Skip to content

Instantly share code, notes, and snippets.

@Zagrebelin
Created February 4, 2016 08:42
Show Gist options
  • Save Zagrebelin/fa1d8f4cfd039bb73813 to your computer and use it in GitHub Desktop.
Save Zagrebelin/fa1d8f4cfd039bb73813 to your computer and use it in GitHub Desktop.
try to race condition
import concurrent.futures
import unittest
from collections import deque, Counter
import time
class QList(list):
def __init__(self, seq=()):
super().__init__(seq)
self._queue = deque()
def free(self, index, set_to=None):
self[index] = set_to
self._queue.append(index)
def append(self, p_object):
if self._queue:
index = self._queue.popleft()
self[index] = p_object
else:
super().append(p_object)
time.sleep(0.01)
index = len(self) - 1
return index
class Foo(unittest.TestCase):
def test_append(self):
fs = []
q = QList()
cnt = 10
workers=2
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
for x in range(cnt):
f = pool.submit(q.append, x)
fs.append(f)
rets = [f._result for f in fs]
c = Counter(rets)
self.assertEqual(cnt, len(c))
for key, value in c.items():
self.assertEqual(1, value)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment