Add _failing_ scheduler test showing one of #612 deadlock reasons

If the lock has too many waiters then some stack overflow raises
RuntimeError and scheduler fails altogether.

ooni/tests/test_scheduler.py

@@ -1,9 +1,10 @@
 import os
+import sys
 import shutil
 import random
 import tempfile
 
-from twisted.internet import defer
+from twisted.internet import reactor, task, defer
 from twisted.trial import unittest
 
 from ooni.agent.scheduler import ScheduledTask, DidNotRun, FileSystemlockAndMutex
@@ -51,8 +52,40 @@ def task(subself):
             self.assertEqual(len(in_file.readlines()), 1)
 
         self.assertEqual(dummy_st.should_run, False)
+        self.assertFalse(os.path.islink(os.path.join(scheduler_directory, identifier + '.lock')))
         shutil.rmtree(scheduler_directory)
 
+    def test_thundering_herd(self):
+        lockno = int(os.getenv('TTH_LOCKNO', str(sys.getrecursionlimit() + 16)))
+        scheduler_directory = tempfile.mkdtemp()
+        counter = os.path.join(scheduler_directory, 'counter')
+        class DummyST(ScheduledTask):
+            @defer.inlineCallbacks
+            def task(subself):
+                self.assertTrue(os.path.islink(os.path.join(scheduler_directory, subself.identifier + '.lock')))
+                with open(counter, 'w+') as fd:
+                    data = fd.read()
+                    fd.seek(0)
+                    if not data:
+                        fd.write('1')
+                        yield task.deferLater(reactor, 1, lambda: 42)
+                    else:
+                        self.assertTrue(False) # should be unreachable due to schedule
+        identifier = "dummy"
+        dummy_st = DummyST(schedule='@daily', identifier=identifier, scheduler_directory=scheduler_directory)
+        dl = defer.DeferredList([dummy_st.run() for i in xrange(lockno)], consumeErrors=True)
+        @dl.addBoth
+        def so_what(results):
+            self.assertFalse(os.path.islink(os.path.join(scheduler_directory, identifier + '.lock')))
+            self.assertEqual(results[0], (True, None)) # do not expect to get `42` here
+            for okflag, ex in results[1:]:
+                self.assertEqual(okflag, False)
+                ex.trap(DidNotRun)
+            self.assertEqual(dummy_st.should_run, False)
+            with open(counter, 'r') as fd:
+                self.assertEqual(fd.read(), '1')
+            shutil.rmtree(scheduler_directory)
+        return dl
 
     @defer.inlineCallbacks
     def test_filesystem_lock_and_mutex(self):