eventlet/tests/queue_test.py

   1 import eventlet
   2 from eventlet import event, hubs, queue
   3 from tests import LimitedTestCase, main
   4
   5
   6 def do_bail(q):
   7     eventlet.Timeout(0, RuntimeError())
   8     try:
   9         result = q.get()
  10         return result
  11     except RuntimeError:
  12         return 'timed out'
  13
  14
  15 class TestQueue(LimitedTestCase):
  16     def test_send_first(self):
  17         q = eventlet.Queue()
  18         q.put('hi')
  19         self.assertEqual(q.get(), 'hi')
  20
  21     def test_send_last(self):
  22         q = eventlet.Queue()
  23
  24         def waiter(q):
  25             self.assertEqual(q.get(), 'hi2')
  26
  27         gt = eventlet.spawn(eventlet.with_timeout, 0.1, waiter, q)
  28         eventlet.sleep(0)
  29         eventlet.sleep(0)
  30         q.put('hi2')
  31         gt.wait()
  32
  33     def test_max_size(self):
  34         q = eventlet.Queue(2)
  35         results = []
  36
  37         def putter(q):
  38             q.put('a')
  39             results.append('a')
  40             q.put('b')
  41             results.append('b')
  42             q.put('c')
  43             results.append('c')
  44
  45         gt = eventlet.spawn(putter, q)
  46         eventlet.sleep(0)
  47         self.assertEqual(results, ['a', 'b'])
  48         self.assertEqual(q.get(), 'a')
  49         eventlet.sleep(0)
  50         self.assertEqual(results, ['a', 'b', 'c'])
  51         self.assertEqual(q.get(), 'b')
  52         self.assertEqual(q.get(), 'c')
  53         gt.wait()
  54
  55     def test_zero_max_size(self):
  56         q = eventlet.Queue(0)
  57
  58         def sender(evt, q):
  59             q.put('hi')
  60             evt.send('done')
  61
  62         def receiver(q):
  63             x = q.get()
  64             return x
  65
  66         evt = event.Event()
  67         gt = eventlet.spawn(sender, evt, q)
  68         eventlet.sleep(0)
  69         assert not evt.ready()
  70         gt2 = eventlet.spawn(receiver, q)
  71         self.assertEqual(gt2.wait(), 'hi')
  72         self.assertEqual(evt.wait(), 'done')
  73         gt.wait()
  74
  75     def test_resize_up(self):
  76         q = eventlet.Queue(0)
  77
  78         def sender(evt, q):
  79             q.put('hi')
  80             evt.send('done')
  81
  82         evt = event.Event()
  83         gt = eventlet.spawn(sender, evt, q)
  84         eventlet.sleep(0)
  85         assert not evt.ready()
  86         q.resize(1)
  87         eventlet.sleep(0)
  88         assert evt.ready()
  89         gt.wait()
  90
  91     def test_resize_down(self):
  92         q = eventlet.Queue(5)
  93
  94         for i in range(5):
  95             q.put(i)
  96
  97         self.assertEqual(list(q.queue), list(range(5)))
  98         q.resize(1)
  99         eventlet.sleep(0)
 100         self.assertEqual(list(q.queue), list(range(5)))
 101
 102     def test_resize_to_Unlimited(self):
 103         q = eventlet.Queue(0)
 104
 105         def sender(evt, q):
 106             q.put('hi')
 107             evt.send('done')
 108
 109         evt = event.Event()
 110         gt = eventlet.spawn(sender, evt, q)
 111         eventlet.sleep()
 112         self.assertFalse(evt.ready())
 113         q.resize(None)
 114         eventlet.sleep()
 115         self.assertTrue(evt.ready())
 116         gt.wait()
 117
 118     def test_multiple_waiters(self):
 119         # tests that multiple waiters get their results back
 120         q = eventlet.Queue()
 121
 122         sendings = ['1', '2', '3', '4']
 123         gts = [eventlet.spawn(q.get) for x in sendings]
 124
 125         eventlet.sleep(0.01)  # get 'em all waiting
 126
 127         q.put(sendings[0])
 128         q.put(sendings[1])
 129         q.put(sendings[2])
 130         q.put(sendings[3])
 131         results = set()
 132         for i, gt in enumerate(gts):
 133             results.add(gt.wait())
 134         self.assertEqual(results, set(sendings))
 135
 136     def test_waiters_that_cancel(self):
 137         q = eventlet.Queue()
 138
 139         gt = eventlet.spawn(do_bail, q)
 140         self.assertEqual(gt.wait(), 'timed out')
 141
 142         q.put('hi')
 143         self.assertEqual(q.get(), 'hi')
 144
 145     def test_getting_before_sending(self):
 146         q = eventlet.Queue()
 147         gt = eventlet.spawn(q.put, 'sent')
 148         self.assertEqual(q.get(), 'sent')
 149         gt.wait()
 150
 151     def test_two_waiters_one_dies(self):
 152         def waiter(q):
 153             return q.get()
 154
 155         q = eventlet.Queue()
 156         dying = eventlet.spawn(do_bail, q)
 157         waiting = eventlet.spawn(waiter, q)
 158         eventlet.sleep(0)
 159         q.put('hi')
 160         self.assertEqual(dying.wait(), 'timed out')
 161         self.assertEqual(waiting.wait(), 'hi')
 162
 163     def test_two_bogus_waiters(self):
 164         q = eventlet.Queue()
 165         gt1 = eventlet.spawn(do_bail, q)
 166         gt2 = eventlet.spawn(do_bail, q)
 167         eventlet.sleep(0)
 168         q.put('sent')
 169         self.assertEqual(gt1.wait(), 'timed out')
 170         self.assertEqual(gt2.wait(), 'timed out')
 171         self.assertEqual(q.get(), 'sent')
 172
 173     def test_waiting(self):
 174         q = eventlet.Queue()
 175         gt1 = eventlet.spawn(q.get)
 176         eventlet.sleep(0)
 177         self.assertEqual(1, q.getting())
 178         q.put('hi')
 179         eventlet.sleep(0)
 180         self.assertEqual(0, q.getting())
 181         self.assertEqual('hi', gt1.wait())
 182         self.assertEqual(0, q.getting())
 183
 184     def test_channel_send(self):
 185         channel = eventlet.Queue(0)
 186         events = []
 187
 188         def another_greenlet():
 189             events.append(channel.get())
 190             events.append(channel.get())
 191
 192         eventlet.spawn(another_greenlet)
 193
 194         events.append('sending')
 195         channel.put('hello')
 196         events.append('sent hello')
 197         channel.put('world')
 198         events.append('sent world')
 199
 200         self.assertEqual(['sending', 'hello', 'sent hello', 'world', 'sent world'], events)
 201
 202     def test_channel_wait(self):
 203         channel = eventlet.Queue(0)
 204         events = []
 205
 206         def another_greenlet():
 207             events.append('sending hello')
 208             channel.put('hello')
 209             events.append('sending world')
 210             channel.put('world')
 211             events.append('sent world')
 212
 213         eventlet.spawn(another_greenlet)
 214
 215         events.append('waiting')
 216         events.append(channel.get())
 217         events.append(channel.get())
 218
 219         self.assertEqual(['waiting', 'sending hello', 'hello', 'sending world', 'world'], events)
 220         eventlet.sleep(0)
 221         self.assertEqual(
 222             ['waiting', 'sending hello', 'hello', 'sending world', 'world', 'sent world'], events)
 223
 224     def test_channel_waiters(self):
 225         c = eventlet.Queue(0)
 226         w1 = eventlet.spawn(c.get)
 227         w2 = eventlet.spawn(c.get)
 228         w3 = eventlet.spawn(c.get)
 229         eventlet.sleep(0)
 230         self.assertEqual(c.getting(), 3)
 231         s1 = eventlet.spawn(c.put, 1)
 232         s2 = eventlet.spawn(c.put, 2)
 233         s3 = eventlet.spawn(c.put, 3)
 234
 235         s1.wait()
 236         s2.wait()
 237         s3.wait()
 238         self.assertEqual(c.getting(), 0)
 239         # NOTE: we don't guarantee that waiters are served in order
 240         results = sorted([w1.wait(), w2.wait(), w3.wait()])
 241         self.assertEqual(results, [1, 2, 3])
 242
 243     def test_channel_sender_timing_out(self):
 244         c = eventlet.Queue(0)
 245         self.assertRaises(queue.Full, c.put, "hi", timeout=0.001)
 246         self.assertRaises(queue.Empty, c.get_nowait)
 247
 248     def test_task_done(self):
 249         channel = queue.Queue(0)
 250         X = object()
 251         gt = eventlet.spawn(channel.put, X)
 252         result = channel.get()
 253         assert result is X, (result, X)
 254         assert channel.unfinished_tasks == 1, channel.unfinished_tasks
 255         channel.task_done()
 256         assert channel.unfinished_tasks == 0, channel.unfinished_tasks
 257         gt.wait()
 258
 259     def test_join_doesnt_block_when_queue_is_already_empty(self):
 260         queue = eventlet.Queue()
 261         queue.join()
 262
 263
 264 def store_result(result, func, *args):
 265     try:
 266         result.append(func(*args))
 267     except Exception as exc:
 268         result.append(exc)
 269
 270
 271 class TestNoWait(LimitedTestCase):
 272     def test_put_nowait_simple(self):
 273         hub = hubs.get_hub()
 274         result = []
 275         q = eventlet.Queue(1)
 276         hub.schedule_call_global(0, store_result, result, q.put_nowait, 2)
 277         hub.schedule_call_global(0, store_result, result, q.put_nowait, 3)
 278         eventlet.sleep(0)
 279         eventlet.sleep(0)
 280         assert len(result) == 2, result
 281         assert result[0] is None, result
 282         assert isinstance(result[1], queue.Full), result
 283
 284     def test_get_nowait_simple(self):
 285         hub = hubs.get_hub()
 286         result = []
 287         q = queue.Queue(1)
 288         q.put(4)
 289         hub.schedule_call_global(0, store_result, result, q.get_nowait)
 290         hub.schedule_call_global(0, store_result, result, q.get_nowait)
 291         eventlet.sleep(0)
 292         assert len(result) == 2, result
 293         assert result[0] == 4, result
 294         assert isinstance(result[1], queue.Empty), result
 295
 296     # get_nowait must work from the mainloop
 297     def test_get_nowait_unlock(self):
 298         hub = hubs.get_hub()
 299         result = []
 300         q = queue.Queue(0)
 301         p = eventlet.spawn(q.put, 5)
 302         assert q.empty(), q
 303         assert q.full(), q
 304         eventlet.sleep(0)
 305         assert q.empty(), q
 306         assert q.full(), q
 307         hub.schedule_call_global(0, store_result, result, q.get_nowait)
 308         eventlet.sleep(0)
 309         assert q.empty(), q
 310         assert q.full(), q
 311         assert result == [5], result
 312         # TODO add ready to greenthread
 313         # assert p.ready(), p
 314         assert p.dead, p
 315         assert q.empty(), q
 316
 317     # put_nowait must work from the mainloop
 318     def test_put_nowait_unlock(self):
 319         hub = hubs.get_hub()
 320         result = []
 321         q = queue.Queue(0)
 322         eventlet.spawn(q.get)
 323         assert q.empty(), q
 324         assert q.full(), q
 325         eventlet.sleep(0)
 326         assert q.empty(), q
 327         assert q.full(), q
 328         hub.schedule_call_global(0, store_result, result, q.put_nowait, 10)
 329         # TODO ready method on greenthread
 330         # assert not p.ready(), p
 331         eventlet.sleep(0)
 332         assert result == [None], result
 333         # TODO ready method
 334         # assert p.ready(), p
 335         assert q.full(), q
 336         assert q.empty(), q
 337
 338
 339 if __name__ == '__main__':
 340     main()