Merge pull request #1051 from aarroyoc/close-stream

Throw exception if closing a closed TCP stream #1046

src/machine/streams.rs

@@ -160,20 +160,6 @@ impl StreamInstance {
     }
 }
 
-impl Drop for StreamInstance {
-    fn drop(&mut self) {
-        match self {
-            StreamInstance::TcpStream(_, ref mut tcp_stream) => {
-                tcp_stream.shutdown(Shutdown::Both).unwrap();
-            }
-            StreamInstance::TlsStream(_, ref mut tls_stream) => {
-                tls_stream.shutdown().unwrap();
-            }
-            _ => {}
-        }
-    }
-}
-
 impl fmt::Debug for StreamInstance {
     fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
         match self {
@@ -596,8 +582,18 @@ impl Stream {
     }
 
     #[inline]
-    pub(crate) fn close(&mut self) {
+    pub(crate) fn close(&mut self) -> Result<(), std::io::Error> {
+        let result = match self.stream_inst.0.borrow_mut().stream_inst {
+            StreamInstance::TcpStream(_, ref mut tcp_stream) => {
+                tcp_stream.shutdown(Shutdown::Both)
+            },
+            StreamInstance::TlsStream(_, ref mut tls_stream) => {
+                tls_stream.shutdown()
+            }
+            _ => Ok(())
+        };
         self.stream_inst.0.borrow_mut().stream_inst = StreamInstance::Null;
+        result
     }
 
     #[inline]

src/machine/system_calls.rs

@@ -2704,11 +2704,29 @@ impl MachineState {
                 }
 
                 if !stream.is_stdin() && !stream.is_stdout() && !stream.is_stderr() {
-                    stream.close();
+                    let close_result = stream.close();
 
                     if let Some(ref alias) = stream.options().alias {
                         indices.stream_aliases.remove(alias);
                     }
+                    if let Err(_) = close_result {
+                        let stub = MachineError::functor_stub(
+                            clause_name!("close"),
+                            1,
+                        );
+
+                        let addr = self.heap.to_unifiable(
+                            HeapCellValue::Stream(stream.clone()),
+                        );
+
+                        return Err(self.error_form(
+                            MachineError::existence_error(
+                                self.heap.h(),
+                                ExistenceError::Stream(addr),
+                            ),
+                            stub,
+                        ));
+                    }
                 }
             }
             &SystemClauseType::CopyToLiftedHeap => match self.store(self.deref(self[temp_v!(1)])) {

tests-pl/issue1046-close-stream-client.pl

@@ -0,0 +1,12 @@
+:- use_module(library(sockets)).
+:- use_module(library(time)).
+:- use_module(library(format)).
+:- use_module(library(charsio)).
+
+test :-
+    Addr = '0.0.0.0',
+    Port = 5000,
+    socket_client_open(Addr:Port, Stream, [type(binary)]),
+    read_line_to_chars(Stream, Line, []),
+    write(Line),
+    close(Stream).

tests-pl/issue1046-close-stream.pl

@@ -0,0 +1,18 @@
+:- use_module(library(sockets)).
+:- use_module(library(time)).
+:- use_module(library(format)).
+
+% Manual test to fix issue 1046
+% Server must be executed first, then client
+% The expected error is an exception thrown here, not a panic
+
+test :-
+    Addr = '0.0.0.0',
+    Port = 5000,
+    socket_server_open(Addr:Port, Socket),
+    format("Listening at port ~d\n", [Port]),
+    socket_server_accept(Socket, _Client, Stream, [type(binary)]),
+    format(Stream, "FIRST\r\n", []),
+    sleep(20),
+    format(Stream, "SECOND\r\n", []),
+    close(Stream).