Re: problem with RPC over SSL and concurrency

On Wed, Mar 18, 2009 at 6:03 PM, Jake Donham <[email protected]> wrote:
> In start_reading, we call nonblock_operation, which queues
> a callback (using Unixqueue.once) that reads from the socket. But in
> cancel_reading, we don't remove the callback. So if we start reading
> and then cancel before the callback is called, it seems to me that the
> read still happens

Here is a patch that fixes the problem for me, by suppressing the
callback if the operation has already been canceled.

Index: uq_ssl.ml
===================================================================
--- uq_ssl.ml   (revision 7383)
+++ uq_ssl.ml   (working copy)
@@ -65,7 +65,7 @@
       failwith "#start_connecting: no longer possible in this state";
     if connecting || accepting then
       failwith "#start_connecting: handshake already in progress";
-    self # nonblock_operation
+    self # nonblock_operation (ref false)
       `Connecting
       (fun () ->
         try
@@ -95,7 +95,7 @@
       failwith "#start_accepting: no longer possible in this state";
     if connecting || accepting then
       failwith "#start_accepting: handshake already in progress";
-    self # nonblock_operation
+    self # nonblock_operation (ref false)
       `Accepting
       (fun () ->
         try
@@ -129,7 +129,8 @@
       failwith "#start_reading: already reading";
     if shutting_down <> None then
       failwith "#start_reading: already shutting down";
-    self # nonblock_operation
+    let canceled = ref false in
+    self # nonblock_operation canceled
       `Reading
       (fun () ->
         try
@@ -158,14 +159,15 @@
               reading <- None;
               (false, false, fun () -> when_done (Some err) 0)
       );
-    reading <- Some when_done
+    reading <- Some (when_done, canceled)


   method cancel_reading () =
     match reading with
       | None ->
          ()
-      | Some f_when_done ->
+      | Some (f_when_done, canceled) ->
+    canceled := true;
          self # cancel_operation `Reading;
          reading <- None;
          f_when_done (Some Uq_engines.Cancelled) 0
@@ -182,7 +184,8 @@
       failwith "#start_writing: already shutting down";
     if wrote_eof then
       failwith "#start_writing: already past EOF";
-    self # nonblock_operation
+    let canceled = ref false in
+    self # nonblock_operation canceled
       `Writing
       (fun () ->
         try
@@ -205,7 +208,7 @@
               writing <- None;
               (false, false, fun () -> when_done (Some err) 0)
       );
-    writing <- Some when_done
+    writing <- Some (when_done, canceled)


   method start_writing_eof ~when_done () =
@@ -216,7 +219,8 @@
     match writing with
       | None ->
          ()
-      | Some f_when_done ->
+      | Some (f_when_done, canceled) ->
+    canceled := true;
          self # cancel_operation `Writing;
          writing <- None;
          f_when_done (Some Uq_engines.Cancelled) 0
@@ -229,7 +233,8 @@
       failwith "#start_shutting_down: still reading or writing";
     if shutting_down <> None then
       failwith "#start_shutting_down: already shutting down";
-    self # nonblock_operation
+    let canceled = ref false in
+    self # nonblock_operation canceled
       `Shutting_down
       (fun () ->
         try
@@ -249,24 +254,28 @@
               shutting_down <- None;
               (false, false, fun () -> when_done (Some err))
       );
-    shutting_down <- Some when_done
+    shutting_down <- Some (when_done, canceled)

   method cancel_shutting_down () =
     match shutting_down with
       | None ->
          ()
-      | Some f_when_done ->
+      | Some (f_when_done, canceled) ->
+    canceled := true;
          self # cancel_operation `Shutting_down;
          shutting_down <- None;
          f_when_done (Some Uq_engines.Cancelled)


-  method private nonblock_operation tag f =
+  method private nonblock_operation canceled tag f =
     Unixqueue.once
       esys
       group
       0.0
       (fun () ->
+        if !canceled
+        then ()
+        else
         let (want_rd, want_wr, action) = f() in
         if want_rd || want_wr then
           pending <- (tag, want_rd, want_wr, f) :: pending;

------------------------------------------------------------------------------
Apps built with the Adobe(R) Flex(R) framework and Flex Builder(TM) are
powering Web 2.0 with engaging, cross-platform capabilities. Quickly and
easily build your RIAs with Flex Builder, the Eclipse(TM)based development
software that enables intelligent coding and step-through debugging.
Download the free 60 day trial. http://p.sf.net/sfu/www-adobe-com