debug: add an option to log synchronous calls

asynq/_debug.pxd

@@ -31,6 +31,7 @@ cdef class DebugOptions(object):
     cdef public bint DUMP_SYNC
     cdef public bint DUMP_STACK
     cdef public bint DUMP_SCHEDULER_STATE
+    cdef public bint DUMP_SYNC_CALLS
 
     cdef public float SCHEDULER_STATE_DUMP_INTERVAL
     cdef public int DEBUG_STR_REPR_MAX_LENGTH

asynq/_debug.py

@@ -40,6 +40,7 @@ def __init__(self):
         self.DUMP_SYNC             = False
         self.DUMP_STACK            = False  # When it's meaningful, e.g. on batch flush
         self.DUMP_SCHEDULER_STATE  = False
+        self.DUMP_SYNC_CALLS       = False
 
         self.SCHEDULER_STATE_DUMP_INTERVAL = 1     # In seconds
         self.DEBUG_STR_REPR_MAX_LENGTH     = 240   # In characters, 0 means infinity

asynq/async_task.py

@@ -73,6 +73,8 @@ def __init__(self, generator, fn, args, kwargs):
             self.creator = asynq.scheduler.get_active_task()
             debug.write('@async: new task: %s, created by %s' %
                 (debug.str(self), debug.str(self.creator)))
+        elif _debug_options.DUMP_SYNC_CALLS:
+            self.creator = asynq.scheduler.get_active_task()
 
     def can_continue(self):
         """Indicates whether this async task has more steps to execute.
@@ -94,6 +96,10 @@ def is_blocked(self):
         return False
 
     def _compute(self):
+        if _debug_options.DUMP_SYNC_CALLS:
+            if self.creator is not None:
+                debug.write('@async: %s called synchronously from %s'
+                    % (debug.str(self), debug.str(self.creator)))
         # Forwards the call to task scheduler
         asynq.scheduler.get_scheduler().await(self)
         # No need to assign a value/error here, since

asynq/debug.py

@@ -41,6 +41,7 @@
 options.DUMP_SYNC             = False
 options.DUMP_STACK            = False  # When it's meaningful, e.g. on batch flush
 options.DUMP_SCHEDULER_STATE  = False
+options.DUMP_SYNC_CALLS       = False
 
 options.SCHEDULER_STATE_DUMP_INTERVAL = 1       # In seconds
 options.DEBUG_STR_REPR_MAX_LENGTH     = 240     # In characters, 0 means infinity

asynq/scheduler.py

@@ -58,6 +58,7 @@ def __init__(self):
     def reset(self):
         self._batches = set()
         self._tasks = []
+        self.active_task = None
 
     def await(self, task):
         """