GSocketInput/OutputStream: fix non-blocking on Windows

The GSocket docs point out that g_socket_send/g_socket_receive may
return G_IO_ERROR_WOULD_BLOCK even if g_socket_condition_check claimed
that they wouldn't. Fix the socket streams to check for that.

https://bugzilla.gnome.org/show_bug.cgi?id=603309

gio/gsocket.c

@@ -2489,6 +2489,14 @@ g_socket_create_source (GSocket      *socket,
  * against the currently-satisfied conditions on @socket. The result
  * is returned.
  *
+ * Note that on Windows, it is possible for an operation to return
+ * %G_IO_ERROR_WOULD_BLOCK even immediately after
+ * g_socket_condition_check() has claimed that the socket is ready for
+ * writing. Rather than calling g_socket_condition_check() and then
+ * writing to the socket if it succeeds, it is generally better to
+ * simply try writing to the socket right away, and try again later if
+ * the initial attempt returns %G_IO_ERROR_WOULD_BLOCK.
+ *
  * It is meaningless to specify %G_IO_ERR or %G_IO_HUP in condition;
  * these conditions will always be set in the output if they are true.
  *

gio/gsocketinputstream.c

@@ -29,6 +29,7 @@
 
 #include <gio/gsimpleasyncresult.h>
 #include <gio/gcancellable.h>
+#include <gio/gioerror.h>
 
 #include "gioalias.h"
 
@@ -48,7 +49,6 @@ struct _GSocketInputStreamPrivate
   /* pending operation metadata */
   GSimpleAsyncResult *result;
   GCancellable *cancellable;
-  gboolean from_mainloop;
   gpointer buffer;
   gsize count;
 };
@@ -125,14 +125,18 @@ g_socket_input_stream_read_ready (GSocket *socket,
   GError *error = NULL;
   gssize result;
 
-  simple = stream->priv->result;
-  stream->priv->result = NULL;
-
   result = g_socket_receive (stream->priv->socket,
 			     stream->priv->buffer,
 			     stream->priv->count,
 			     stream->priv->cancellable,
 			     &error);
+
+  if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
+    return TRUE;
+
+  simple = stream->priv->result;
+  stream->priv->result = NULL;
+
   if (result >= 0)
     g_simple_async_result_set_op_res_gssize (simple, result);
 
@@ -145,11 +149,7 @@ g_socket_input_stream_read_ready (GSocket *socket,
   if (stream->priv->cancellable)
     g_object_unref (stream->priv->cancellable);
 
-  if (stream->priv->from_mainloop)
+  g_simple_async_result_complete (simple);
-    g_simple_async_result_complete (simple);
-  else
-    g_simple_async_result_complete_in_idle (simple);
-
   g_object_unref (simple);
 
   return FALSE;
@@ -165,6 +165,7 @@ g_socket_input_stream_read_async (GInputStream        *stream,
                                   gpointer             user_data)
 {
   GSocketInputStream *input_stream = G_SOCKET_INPUT_STREAM (stream);
+  GSource *source;
 
   g_assert (input_stream->priv->result == NULL);
 
@@ -177,25 +178,14 @@ g_socket_input_stream_read_async (GInputStream        *stream,
   input_stream->priv->buffer = buffer;
   input_stream->priv->count = count;
 
-  if (!g_socket_condition_check (input_stream->priv->socket, G_IO_IN))
+  source = g_socket_create_source (input_stream->priv->socket,
-    {
+				   G_IO_IN | G_IO_HUP | G_IO_ERR,
-      GSource *source;
+				   cancellable);
-
+  g_source_set_callback (source,
-      input_stream->priv->from_mainloop = TRUE;
+			 (GSourceFunc) g_socket_input_stream_read_ready,
-      source = g_socket_create_source (input_stream->priv->socket,
+			 g_object_ref (input_stream), g_object_unref);
-                                       G_IO_IN | G_IO_HUP | G_IO_ERR,
+  g_source_attach (source, g_main_context_get_thread_default ());
-                                       cancellable);
+  g_source_unref (source);
-      g_source_set_callback (source,
-                             (GSourceFunc) g_socket_input_stream_read_ready,
-                             g_object_ref (input_stream), g_object_unref);
-      g_source_attach (source, g_main_context_get_thread_default ());
-      g_source_unref (source);
-    }
-  else
-    {
-      input_stream->priv->from_mainloop = FALSE;
-      g_socket_input_stream_read_ready (input_stream->priv->socket, G_IO_IN, input_stream);
-    }
 }
 
 static gssize

gio/gsocketoutputstream.c

@@ -30,6 +30,7 @@
 
 #include <gio/gsimpleasyncresult.h>
 #include <gio/gcancellable.h>
+#include <gio/gioerror.h>
 #include "glibintl.h"
 
 #include "gioalias.h"
@@ -50,7 +51,6 @@ struct _GSocketOutputStreamPrivate
   /* pending operation metadata */
   GSimpleAsyncResult *result;
   GCancellable *cancellable;
-  gboolean from_mainloop;
   gconstpointer buffer;
   gsize count;
 };
@@ -127,14 +127,18 @@ g_socket_output_stream_write_ready (GSocket *socket,
   GError *error = NULL;
   gssize result;
 
-  simple = stream->priv->result;
-  stream->priv->result = NULL;
-
   result = g_socket_send (stream->priv->socket,
 			  stream->priv->buffer,
 			  stream->priv->count,
 			  stream->priv->cancellable,
 			  &error);
+
+  if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
+    return TRUE;
+
+  simple = stream->priv->result;
+  stream->priv->result = NULL;
+
   if (result >= 0)
     g_simple_async_result_set_op_res_gssize (simple, result);
 
@@ -147,11 +151,7 @@ g_socket_output_stream_write_ready (GSocket *socket,
   if (stream->priv->cancellable)
     g_object_unref (stream->priv->cancellable);
 
-  if (stream->priv->from_mainloop)
+  g_simple_async_result_complete (simple);
-    g_simple_async_result_complete (simple);
-  else
-    g_simple_async_result_complete_in_idle (simple);
-
   g_object_unref (simple);
 
   return FALSE;
@@ -167,6 +167,7 @@ g_socket_output_stream_write_async (GOutputStream        *stream,
                                     gpointer              user_data)
 {
   GSocketOutputStream *output_stream = G_SOCKET_OUTPUT_STREAM (stream);
+  GSource *source;
 
   g_assert (output_stream->priv->result == NULL);
 
@@ -179,25 +180,14 @@ g_socket_output_stream_write_async (GOutputStream        *stream,
   output_stream->priv->buffer = buffer;
   output_stream->priv->count = count;
 
-  if (!g_socket_condition_check (output_stream->priv->socket, G_IO_OUT))
+  source = g_socket_create_source (output_stream->priv->socket,
-    {
+				   G_IO_OUT | G_IO_HUP | G_IO_ERR,
-      GSource *source;
+				   cancellable);
-
+  g_source_set_callback (source,
-      output_stream->priv->from_mainloop = TRUE;
+			 (GSourceFunc) g_socket_output_stream_write_ready,
-      source = g_socket_create_source (output_stream->priv->socket,
+			 g_object_ref (output_stream), g_object_unref);
-                                       G_IO_OUT | G_IO_HUP | G_IO_ERR,
+  g_source_attach (source, g_main_context_get_thread_default ());
-                                       cancellable);
+  g_source_unref (source);
-      g_source_set_callback (source,
-                             (GSourceFunc) g_socket_output_stream_write_ready,
-                             g_object_ref (output_stream), g_object_unref);
-      g_source_attach (source, g_main_context_get_thread_default ());
-      g_source_unref (source);
-    }
-  else
-    {
-      output_stream->priv->from_mainloop = FALSE;
-      g_socket_output_stream_write_ready (output_stream->priv->socket, G_IO_OUT, output_stream);
-    }
 }
 
 static gssize