microbenchmarks and timing (#67)

src/orchpylib.cc

@@ -550,9 +550,7 @@ PyObject* deserialize_call(PyObject* self, PyObject* args) {
   }
   Worker* worker;
   PyObjectToWorker(worker_capsule, &worker);
-  if (objrefs.size() > 0) {
-    worker->decrement_reference_count(objrefs);
-  }
+  worker->decrement_reference_count(objrefs);
   int resultsize = call->result_size();
   std::vector<ObjRef> result_objrefs;
   PyObject* resultlist = PyList_New(resultsize);

src/worker.cc

@@ -215,14 +215,16 @@ void Worker::increment_reference_count(std::vector<ObjRef> &objrefs) {
     ORCH_LOG(ORCH_DEBUG, "Attempting to increment_reference_count for objrefs, but connected_ = " << connected_ << " so returning instead.");
     return;
   }
-  ClientContext context;
-  IncrementRefCountRequest request;
-  for (int i = 0; i < objrefs.size(); ++i) {
-    ORCH_LOG(ORCH_REFCOUNT, "Incrementing reference count for objref " << objrefs[i]);
-    request.add_objref(objrefs[i]);
+  if (objrefs.size() > 0) {
+    ClientContext context;
+    IncrementRefCountRequest request;
+    for (int i = 0; i < objrefs.size(); ++i) {
+      ORCH_LOG(ORCH_REFCOUNT, "Incrementing reference count for objref " << objrefs[i]);
+      request.add_objref(objrefs[i]);
+    }
+    AckReply reply;
+    scheduler_stub_->IncrementRefCount(&context, request, &reply);
   }
-  AckReply reply;
-  scheduler_stub_->IncrementRefCount(&context, request, &reply);
 }
 
 void Worker::decrement_reference_count(std::vector<ObjRef> &objrefs) {
@@ -230,14 +232,16 @@ void Worker::decrement_reference_count(std::vector<ObjRef> &objrefs) {
     ORCH_LOG(ORCH_DEBUG, "Attempting to decrement_reference_count, but connected_ = " << connected_ << " so returning instead.");
     return;
   }
-  ClientContext context;
-  DecrementRefCountRequest request;
-  for (int i = 0; i < objrefs.size(); ++i) {
-    ORCH_LOG(ORCH_REFCOUNT, "Decrementing reference count for objref " << objrefs[i]);
-    request.add_objref(objrefs[i]);
+  if (objrefs.size() > 0) {
+    ClientContext context;
+    DecrementRefCountRequest request;
+    for (int i = 0; i < objrefs.size(); ++i) {
+      ORCH_LOG(ORCH_REFCOUNT, "Decrementing reference count for objref " << objrefs[i]);
+      request.add_objref(objrefs[i]);
+    }
+    AckReply reply;
+    scheduler_stub_->DecrementRefCount(&context, request, &reply);
   }
-  AckReply reply;
-  scheduler_stub_->DecrementRefCount(&context, request, &reply);
 }
 
 void Worker::register_function(const std::string& name, size_t num_return_vals) {

test/microbenchmarks.py

@@ -0,0 +1,85 @@
+import unittest
+import orchpy
+import orchpy.services as services
+import time
+import os
+import numpy as np
+
+import test_functions
+
+class MicroBenchmarkTest(unittest.TestCase):
+
+  def testTiming(self):
+    test_dir = os.path.dirname(os.path.abspath(__file__))
+    test_path = os.path.join(test_dir, "testrecv.py")
+    services.start_singlenode_cluster(return_drivers=False, num_workers_per_objstore=3, worker_path=test_path)
+
+    # measure the time required to submit a remote call to the scheduler
+    elapsed_times = []
+    for _ in range(1000):
+      start_time = time.time()
+      test_functions.empty_function()
+      end_time = time.time()
+      elapsed_times.append(end_time - start_time)
+    elapsed_times = np.sort(elapsed_times)
+    average_elapsed_time = sum(elapsed_times) / 1000
+    print "Time required to submit an empty function call:"
+    print "    Average: {}".format(average_elapsed_time)
+    print "    90th percentile: {}".format(elapsed_times[900])
+    print "    99th percentile: {}".format(elapsed_times[990])
+    print "    worst:           {}".format(elapsed_times[999])
+    self.assertTrue(average_elapsed_time < 0.00038)
+
+    # measure the time required to submit a remote call to the scheduler (where the remote call returns one value)
+    elapsed_times = []
+    for _ in range(1000):
+      start_time = time.time()
+      test_functions.trivial_function()
+      end_time = time.time()
+      elapsed_times.append(end_time - start_time)
+    elapsed_times = np.sort(elapsed_times)
+    average_elapsed_time = sum(elapsed_times) / 1000
+    print "Time required to submit a trivial function call:"
+    print "    Average: {}".format(average_elapsed_time)
+    print "    90th percentile: {}".format(elapsed_times[900])
+    print "    99th percentile: {}".format(elapsed_times[990])
+    print "    worst:           {}".format(elapsed_times[999])
+    self.assertTrue(average_elapsed_time < 0.001)
+
+    # measure the time required to submit a remote call to the scheduler and pull the result
+    elapsed_times = []
+    for _ in range(1000):
+      start_time = time.time()
+      x = test_functions.trivial_function()
+      orchpy.pull(x)
+      end_time = time.time()
+      elapsed_times.append(end_time - start_time)
+    elapsed_times = np.sort(elapsed_times)
+    average_elapsed_time = sum(elapsed_times) / 1000
+    print "Time required to submit a trivial function call and pull the result:"
+    print "    Average: {}".format(average_elapsed_time)
+    print "    90th percentile: {}".format(elapsed_times[900])
+    print "    99th percentile: {}".format(elapsed_times[990])
+    print "    worst:           {}".format(elapsed_times[999])
+    self.assertTrue(average_elapsed_time < 0.0013)
+
+    # measure the time required to do do a push
+    elapsed_times = []
+    for _ in range(1000):
+      start_time = time.time()
+      orchpy.push(1)
+      end_time = time.time()
+      elapsed_times.append(end_time - start_time)
+    elapsed_times = np.sort(elapsed_times)
+    average_elapsed_time = sum(elapsed_times) / 1000
+    print "Time required to push an int:"
+    print "    Average: {}".format(average_elapsed_time)
+    print "    90th percentile: {}".format(elapsed_times[900])
+    print "    99th percentile: {}".format(elapsed_times[990])
+    print "    worst:           {}".format(elapsed_times[999])
+    self.assertTrue(average_elapsed_time < 0.00087)
+
+    services.cleanup()
+
+if __name__ == '__main__':
+    unittest.main()

test/test_functions.py

@@ -29,4 +29,12 @@ def test_alias_g():
 def test_alias_h():
   return test_alias_g()
 
-# Test reference counting
+# Test timing
+
+@orchpy.distributed([], [])
+def empty_function():
+  return ()
+
+@orchpy.distributed([], [int])
+def trivial_function():
+  return 1