Hi friends of VIFF,

I've finally completed the patch for a two-threaded VIFF where most of the VIFF code runs in a separate thread. The patch is against the tip of my repository: http://hg.viff.dk/mkeller


It turned out be not so straight-forward as I thought. I had to use a recursion, as in the hack, but I refined it to ensure that the recursion limit isn't exceeded.

Benchmarks:
Unfortunately, this solution is slower than the hack, e.g. one AES block encryption takes 4 seconds compared to 3 seconds with the hack. On the other hand, the preprocessing time in the actively secure multiplication is linear and not quadratic, whereas the online time is significantly larger:

        two-threaded            hack                    original
(n,t)   online  preprocessing   online  preprocessing   online  preproc.
(4,1)   6       22              4       17              4       20
(7,2)   10      37              6       29              6       42
(10,3)  13      53              8       42              8       82
(13,4)  17      68              10      56              10      136
(16,5)  20      84              12      68              12      208
(19,6)  23      106             13      83              14      287
(22,7)  26      120             15      98              17      377

I did some profiling and didn't find an obvious reason why the two-thread is slower. Therefore, I guess that the reason is the multi-threading implementation of Python (which could be better, as mentioned in the discussion about the hack). The guess is also supported by the fact that having an own thread for every callback, which I also tried, turned out to be really slow.

Unit tests:
All unit test get passed, even the previosly skipped test_multiple_callbacks. This because I added the @increment_pc decorator to schedule_callback(). This of course changes the program counters heavily but I didn't experience any problems.

Best regards,
Marcel
diff -r e2759515f57f apps/aes.py
--- a/apps/aes.py	Thu Mar 05 21:02:57 2009 +0100
+++ b/apps/aes.py	Tue Apr 21 17:25:45 2009 +0200
@@ -82,7 +82,7 @@
         rt.shutdown()
 
     g = gather_shares(opened_ciphertext)
-    g.addCallback(fin)
+    rt.schedule_callback(g, fin)
 
 def share_key(rt):
     key =  []
diff -r e2759515f57f apps/benchmark.py
--- a/apps/benchmark.py	Thu Mar 05 21:02:57 2009 +0100
+++ b/apps/benchmark.py	Tue Apr 21 17:25:45 2009 +0200
@@ -91,6 +91,7 @@
     print "Total time used: %.3f sec" % (stop-start)
     print "Time per %s operation: %.0f ms" % (what, 1000*(stop-start) / count)
     print "*" * 6
+    sys.stdout.flush()
 
 
 operations = ["mul", "compToft05", "compToft07", "eq"]
@@ -174,7 +175,7 @@
             # run with no preprocessing. So they are quite brittle.
             if self.operation == operator.mul:
                 key = ("generate_triples", (Zp,))
-                desc = [(i, 1, 0) for i in range(2, 2 + count)]
+                desc = [(2, 2, 2 * count + 1, 2, i, 1, 0) for i in range(1, count + 1)]
                 program_desc.setdefault(key, []).extend(desc)
             elif isinstance(self.rt, ComparisonToft05Mixin):
                 key = ("generate_triples", (GF256,))
@@ -228,7 +229,8 @@
         if seconds > 0:
             print "Starting test in %d" % seconds
             sys.stdout.flush()
-            reactor.callLater(1, self.countdown, None, seconds - 1)
+            time.sleep(1)
+            self.countdown(None, seconds - 1)
         else:
             print "Starting test now"
             sys.stdout.flush()
@@ -255,6 +257,7 @@
     def run_test(self, _):
         c_shares = []
         record_start("parallel test")
+        sys.stdout.flush()
         while self.a_shares and self.b_shares:
             a = self.a_shares.pop()
             b = self.b_shares.pop()
diff -r e2759515f57f apps/millionaires.py
--- a/apps/millionaires.py	Thu Mar 05 21:02:57 2009 +0100
+++ b/apps/millionaires.py	Tue Apr 21 17:25:45 2009 +0200
@@ -97,10 +97,10 @@
         # the argument (which is None since self.results_ready does
         # not return anything), so we throw it away using a lambda
         # expressions which ignores its first argument.
-        results.addCallback(lambda _: runtime.synchronize())
+        runtime.schedule_callback(results, lambda _: runtime.synchronize())
         # The next callback shuts the runtime down, killing the
         # connections between the players.
-        results.addCallback(lambda _: runtime.shutdown())
+        runtime.schedule_callback(results, lambda _: runtime.shutdown())
 
     def results_ready(self, results):
         # Since this method is called as a callback above, the results
diff -r e2759515f57f viff/active.py
--- a/viff/active.py	Thu Mar 05 21:02:57 2009 +0100
+++ b/viff/active.py	Tue Apr 21 17:25:45 2009 +0200
@@ -501,6 +501,7 @@
         result = Share(self, share_x.field)
         # This is the Deferred we will do processing on.
         triple = self.get_triple(share_x.field)
+        triple.addCallback(gather_shares)
         self.schedule_callback(triple, finish_mul)
         # We add the result to the chains in triple.
         triple.chainDeferred(result)
diff -r e2759515f57f viff/aes.py
--- a/viff/aes.py	Thu Mar 05 21:02:57 2009 +0100
+++ b/viff/aes.py	Tue Apr 21 17:25:45 2009 +0200
@@ -374,9 +374,9 @@
                 trigger.addCallback(progress, i, time.time())
 
                 if (i < self.rounds - 1):
-                    self.runtime.schedule_callback(trigger, round, state, i + 1)
+                    self.runtime.schedule_complex_callback(trigger, round, state, i + 1)
                 else:
-                    self.runtime.schedule_callback(trigger, final_round, state)
+                    self.runtime.schedule_complex_callback(trigger, final_round, state)
 
             prep_progress(i, start_round)
 
diff -r e2759515f57f viff/equality.py
--- a/viff/equality.py	Thu Mar 05 21:02:57 2009 +0100
+++ b/viff/equality.py	Tue Apr 21 17:25:45 2009 +0200
@@ -74,7 +74,7 @@
                 xj = (-1) * (1/Zp(2)) * (bj - 1)
             return xj
 
-        x = [cj.addCallback(finish, bj) for cj, bj in zip(c, b)]
+        x = [self.schedule_callback(cj, finish, bj) for cj, bj in zip(c, b)]
 
         # Take the product (this is here the same as the "and") of all
         # the x'es
diff -r e2759515f57f viff/passive.py
--- a/viff/passive.py	Thu Mar 05 21:02:57 2009 +0100
+++ b/viff/passive.py	Tue Apr 21 17:25:45 2009 +0200
@@ -98,7 +98,7 @@
                         d = Share(self, share.field, (share.field(peer_id), share))
                     else:
                         d = self._expect_share(peer_id, share.field)
-                        self.schedule_callback(d, lambda s, peer_id: (s.field(peer_id), s), peer_id)
+                        d.addCallback(lambda s, peer_id: (s.field(peer_id), s), peer_id)
                     deferreds.append(d)
                 return recombine(deferreds)
 
diff -r e2759515f57f viff/runtime.py
--- a/viff/runtime.py	Thu Mar 05 21:02:57 2009 +0100
+++ b/viff/runtime.py	Tue Apr 21 17:25:45 2009 +0200
@@ -39,9 +39,10 @@
 from collections import deque
 
 from viff.field import GF256, FieldElement
-from viff.util import wrapper, rand, deep_wait, track_memory_usage
+from viff.util import wrapper, rand, deep_wait, track_memory_usage, \
+     clone_deferred
 
-from twisted.internet import reactor
+from twisted.internet import reactor, selectreactor
 from twisted.internet.task import LoopingCall
 from twisted.internet.error import ConnectionDone, CannotListenError
 from twisted.internet.defer import Deferred, DeferredList, gatherResults, succeed
@@ -49,6 +50,10 @@
 from twisted.internet.protocol import ReconnectingClientFactory, ServerFactory
 from twisted.protocols.basic import Int16StringReceiver
 
+from Queue import Queue, Empty
+from threading import Lock, Event
+import sys
+
 
 class Share(Deferred):
     """A shared number.
@@ -78,6 +83,7 @@
         self.field = field
         if value is not None:
             self.callback(value)
+        self.priority = 0
 
     def __add__(self, other):
         """Addition."""
@@ -212,14 +218,19 @@
 
         for index, share in enumerate(shares):
             share.addCallbacks(self._callback_fired, self._callback_fired,
-                               callbackArgs=(index, True),
+                               callbackArgs=(index, True, share.priority),
                                errbackArgs=(index, False))
 
-    def _callback_fired(self, result, index, success):
+    def _callback_fired(self, result, index, success, priority=0):
         self.results[index] = (success, result)
         self.missing_shares -= 1
         if not self.called and self.missing_shares == 0:
-            self.callback(self.results)
+            if self.priority >= priority:
+                self.callback(self.results)
+            else:
+                self.pause()
+                self.callback(self.results)
+                self.runtime.deferred_queue.put((self, None))
         return result
 
 
@@ -266,6 +277,9 @@
         self.lost_connection = Deferred()
         #: Data expected to be received in the future.
         self.incoming_data = {}
+        #: Lock to protect :attr:`incoming_data`.
+        self.incoming_lock = Lock()
+        self.activation_counter = 0
 
     def connectionMade(self):
         self.sendString(str(self.factory.runtime.id))
@@ -301,21 +315,32 @@
             program_counter, data_type, data = marshal.loads(string)
             key = (program_counter, data_type)
 
+            self.incoming_lock.acquire()
             deq = self.incoming_data.setdefault(key, deque())
             if deq and isinstance(deq[0], Deferred):
                 deferred = deq.popleft()
                 if not deq:
                     del self.incoming_data[key]
+                self.incoming_lock.release()
+                # queue deferred
+                deferred.pause()
                 deferred.callback(data)
+                self.factory.runtime.deferred_queue.put((deferred, program_counter))
             else:
                 deq.append(data)
+                self.incoming_lock.release()
 
             # TODO: marshal.loads can raise EOFError, ValueError, and
             # TypeError. They should be handled somehow.
 
     def sendData(self, program_counter, data_type, data):
         send_data = (program_counter, data_type, data)
-        self.sendString(marshal.dumps(send_data))
+        reactor.threadCallQueue.append((self.sendString, [marshal.dumps(send_data)], {}))
+        self.activation_counter +=1
+
+        if (self.activation_counter >= 1):
+            reactor.wakeUp()
+            self.activation_counter = 0
 
     def sendShare(self, program_counter, share):
         """Send a share.
@@ -501,6 +526,15 @@
         # communicating with ourselves.
         self.add_player(player, None)
 
+        #: Blocking queue for Deferreds.
+        self.deferred_queue = Queue(0)
+        #: Current recursion depth.
+        self.recursion_depth = 0
+        #: Activation counter.
+        self.activation_counter = 0
+        #: Event is set while the reactor is waiting (select syscall).
+        self.select_event = Event()
+
     def add_player(self, player, protocol):
         self.players[player.id] = player
         self.num_players = len(self.players)
@@ -522,18 +556,18 @@
             results = [maybeDeferred(self.port.stopListening)]
             for protocol in self.protocols.itervalues():
                 results.append(protocol.lost_connection)
-                protocol.loseConnection()
+                reactor.callFromThread(protocol.loseConnection)
             return DeferredList(results)
 
-        def stop_reactor(_):
+        def stop_reactor(_, self):
             print "done."
             print "Stopping reactor... ",
-            reactor.stop()
+            reactor.callFromThread(reactor.stop)
             print "done."
 
         sync = self.synchronize()
         sync.addCallback(close_connections)
-        sync.addCallback(stop_reactor)
+        sync.addCallback(stop_reactor, self)
         return sync
 
     def wait_for(self, *vars):
@@ -542,8 +576,9 @@
         The runtime is shut down when all variables are calculated.
         """
         dl = DeferredList(vars)
-        dl.addCallback(lambda _: self.shutdown())
+        self.schedule_callback(dl, lambda _: self.shutdown())
 
+    @increment_pc
     def schedule_callback(self, deferred, func, *args, **kwargs):
         """Schedule a callback on a deferred with the correct program
         counter.
@@ -574,7 +609,64 @@
             finally:
                 self.program_counter[:] = current_pc
 
-        deferred.addCallback(callback_wrapper, *args, **kwargs)
+        return deferred.addCallback(callback_wrapper, *args, **kwargs)
+
+    def schedule_complex_callback(self, share, func, *args, **kwargs):
+        """Schedule a complex callback, i.e. a callback which blocks a
+        long time."""
+
+        assert isinstance(share, Share), "Only shares can have complex callbacks."
+
+        share.priority = -1
+        self.schedule_callback(share, func, *args, **kwargs)
+
+    def profile_deferred_queue_loop(self):
+        import cProfile
+        prof = cProfile.Profile()
+        prof.runcall(self.deferred_queue_loop)
+        prof.print_stats(1)
+
+    def deferred_queue_loop(self):
+        while True:
+            deferred, pc = self.deferred_queue.get()
+
+            if deferred is not None:
+                deferred.unpause()
+                from twisted.python import failure
+                if isinstance(deferred.result,failure.Failure):
+                    deferred.result.printTraceback()
+                sys.stdout.flush()
+            else:
+                return
+
+    def process_deferred_queue(self):
+        self.select_event.wait()
+
+        max_depth = 1
+
+        if (self.recursion_depth >= max_depth):
+            return
+
+        self.recursion_depth += 1
+
+        while True:
+            try:
+                deferred, pc = self.deferred_queue.get(block=False)
+            except Empty:
+                break
+            else:
+                if deferred is not None:
+                    if isinstance(deferred, Share) and \
+                           self.recursion_depth / float(max_depth) - deferred.priority > 1:
+                        self.deferred_queue.put((deferred, None))
+                        break
+                    else:
+                        deferred.unpause()
+                else:
+                    self.deferred_queue.put((None, None))
+                    break
+
+        self.recursion_depth -= 1
 
     @increment_pc
     def synchronize(self):
@@ -598,16 +690,25 @@
         pc = tuple(self.program_counter)
         key = (pc, data_type)
 
+        lock = self.protocols[peer_id].incoming_lock
+        lock.acquire()
         deq = self.protocols[peer_id].incoming_data.setdefault(key, deque())
         if deq and not isinstance(deq[0], Deferred):
             # We have already received some data from the other side.
             data = deq.popleft()
             if not deq:
                 del self.protocols[peer_id].incoming_data[key]
+            lock.release()
             deferred.callback(data)
         else:
             # We have not yet received anything from the other side.
             deq.append(deferred)
+            lock.release()
+            self.activation_counter += 1
+
+            if (self.activation_counter >= 1):
+                self.process_deferred_queue()
+                self.activation_counter = 0
 
     def _exchange_shares(self, peer_id, field_element):
         """Exchange shares with another player.
@@ -789,24 +890,13 @@
         # profiler here and stop it upon shutdown, but this triggers
         # http://bugs.python.org/issue1375 since the start and stop
         # calls are in different stack frames.
-        import hotshot
-        prof = hotshot.Profile("player-%d.prof" % id)
+        import cProfile
+        prof = cProfile.Profile()
         old_run = reactor.run
         def new_run(*args, **kwargs):
             print "Starting reactor with profiling"
             prof.runcall(old_run, *args, **kwargs)
-
-            import sys
-            import hotshot.stats
-            print "Loading profiling statistics...",
-            sys.stdout.flush()
-            stats = hotshot.stats.load("player-%d.prof" % id)
-            print "done."
-            print
-            stats.strip_dirs()
-            stats.sort_stats("time", "calls")
-            stats.print_stats(40)
-            stats.dump_stats("player-%d.pstats" % id)
+            prof.print_stats(1)
         reactor.run = new_run
 
     # This will yield a Runtime when all protocols are connected.
@@ -874,7 +964,38 @@
             print "Will connect to %s" % player
             connect(player.host, player.port)
 
-    return result
+    # Start the main callback in the separate thread.
+    deferred_runtime = clone_deferred(result)
+    deferred_runtime.pause()
+
+    def start_deferred_queue_loop(runtime, deferred_runtime, options):
+        print "Start VIFF thread"
+
+        runtime.deferred_queue.put((deferred_runtime, None))
+
+        if options and options.profile:
+            reactor.callInThread(runtime.profile_deferred_queue_loop)
+        else:
+            reactor.callInThread(runtime.deferred_queue_loop)
+
+        reactor.addSystemEventTrigger("before", "shutdown",
+                                      runtime.deferred_queue.put, (None, None))
+        return runtime
+
+    result.addCallback(start_deferred_queue_loop, deferred_runtime, options)
+
+    # Monkey patch selectreactor to known when it is waiting.
+    plain_select = selectreactor._select
+
+    def patched_select(*args, **kwargs):
+        runtime.select_event.set()
+        result = plain_select(*args, **kwargs)
+        runtime.select_event.clear()
+        return result
+
+    selectreactor._select = patched_select
+
+    return deferred_runtime
 
 if __name__ == "__main__":
     import doctest    #pragma NO COVER
diff -r e2759515f57f viff/test/test_basic_runtime.py
--- a/viff/test/test_basic_runtime.py	Thu Mar 05 21:02:57 2009 +0100
+++ b/viff/test/test_basic_runtime.py	Tue Apr 21 17:25:45 2009 +0200
@@ -62,13 +62,13 @@
         """
 
         def verify_program_counter(_):
-            self.assertEquals(runtime.program_counter, [0])
+            self.assertEquals(runtime.program_counter, [1, 0])
 
         d = Deferred()
         runtime.schedule_callback(d, verify_program_counter)
 
         runtime.synchronize()
-        self.assertEquals(runtime.program_counter, [1])
+        self.assertEquals(runtime.program_counter, [2])
 
         # Now trigger verify_program_counter.
         d.callback(None)
@@ -129,8 +129,6 @@
         d2.callback(None)
 
         return gatherResults([d1, d2])
-    test_multiple_callbacks.skip = ("TODO: Scheduling callbacks fails to "
-                                    "increment program counter!")
 
     @protocol
     def test_multi_send(self, runtime):
diff -r e2759515f57f viff/test/util.py
--- a/viff/test/util.py	Thu Mar 05 21:02:57 2009 +0100
+++ b/viff/test/util.py	Tue Apr 21 17:25:45 2009 +0200
@@ -19,16 +19,19 @@
 
 from twisted.internet.defer import Deferred, gatherResults, maybeDeferred
 from twisted.trial.unittest import TestCase
+from twisted.internet import reactor
 
 from viff.passive import PassiveRuntime
 from viff.runtime import Share, ShareExchanger, ShareExchangerFactory
 from viff.field import GF
 from viff.config import generate_configs, load_config
-from viff.util import rand
+from viff.util import rand, clone_deferred
 from viff.test.loopback import loopbackAsync
 
 from random import Random
 
+from threading import Thread
+
 
 def protocol(method):
     """Decorator for protocol tests.
@@ -44,9 +47,12 @@
     def wrapper(self):
 
         def shutdown_protocols(result, runtime):
+            # Stop VIFF thread.
+            runtime.deferred_queue.put((None, None))
             # TODO: this should use runtime.shutdown instead.
             for protocol in runtime.protocols.itervalues():
-                protocol.loseConnection()
+                reactor.callFromThread(protocol.loseConnection)
+            reactor.wakeUp()
             # If we were called as an errback, then returning the
             # result signals the original error to Trial.
             return result
@@ -127,6 +133,7 @@
         self.close_sentinels = []
 
         self.runtimes = []
+        self.real_runtimes = []
         for id in reversed(range(1, self.num_players+1)):
             _, players = load_config(configs[id])
             self.create_loopback_runtime(id, players)
@@ -143,6 +150,9 @@
         for protocol in self.protocols.itervalues():
             protocol.transport.close()
 
+        for runtime in self.real_runtimes:
+            runtime.deferred_queue.put((None, None))
+
     def create_loopback_runtime(self, id, players):
         """Create a L{Runtime} connected with a loopback.
 
@@ -163,7 +173,10 @@
         # We add the Deferred passed to ShareExchangerFactory and not
         # the Runtime, since we want everybody to wait until all
         # runtimes are ready.
-        self.runtimes.append(result)
+        deferred_runtime = clone_deferred(result)
+        deferred_runtime.pause()
+        self.runtimes.append(deferred_runtime)
+        self.real_runtimes.append(runtime)
 
         for peer_id in players:
             if peer_id != id:
@@ -189,6 +202,14 @@
                     sentinel = loopbackAsync(server, client)
                     self.close_sentinels.append(sentinel)
 
+        def start_deferred_queue_loop(runtime, deferred_runtime):
+            runtime.deferred_queue.put((deferred_runtime, None))
+            runtime.select_event.set()
+            reactor.callInThread(runtime.deferred_queue_loop)
+            return runtime
+
+        result.addCallback(start_deferred_queue_loop, deferred_runtime)
+
 
 class BinaryOperatorTestCase:
     """Test a binary operator.
_______________________________________________
viff-devel mailing list (http://viff.dk/)
viff-devel@viff.dk
http://lists.viff.dk/listinfo.cgi/viff-devel-viff.dk

Reply via email to