diff --git a/src/orchpylib.cc b/src/orchpylib.cc index 58d8bccc4..d408b3ce9 100644 --- a/src/orchpylib.cc +++ b/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 result_objrefs; PyObject* resultlist = PyList_New(resultsize); diff --git a/src/worker.cc b/src/worker.cc index 0c4c9f03e..d1b0373ff 100644 --- a/src/worker.cc +++ b/src/worker.cc @@ -215,14 +215,16 @@ void Worker::increment_reference_count(std::vector &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 &objrefs) { @@ -230,14 +232,16 @@ void Worker::decrement_reference_count(std::vector &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) { diff --git a/test/microbenchmarks.py b/test/microbenchmarks.py new file mode 100644 index 000000000..b6406bd00 --- /dev/null +++ b/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() diff --git a/test/test_functions.py b/test/test_functions.py index 7f6ef43e2..935fef283 100644 --- a/test/test_functions.py +++ b/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