- Construct matrices of size N x D - Start K threads - Push all N rows into the queue of capacity C - Pull all N rows out of the queue. - Verify that the output matrices are permutation of the rows of the original matrices.
(self, num_threads, num_elements,
capacity, num_blobs, do)
| 1294 | do=st.sampled_from(hu.device_options)) |
| 1295 | @settings(deadline=10000) |
| 1296 | def test_blobs_queue_threading(self, num_threads, num_elements, |
| 1297 | capacity, num_blobs, do): |
| 1298 | """ |
| 1299 | - Construct matrices of size N x D |
| 1300 | - Start K threads |
| 1301 | - Push all N rows into the queue of capacity C |
| 1302 | - Pull all N rows out of the queue. |
| 1303 | - Verify that the output matrices are permutation of the rows of the |
| 1304 | original matrices. |
| 1305 | """ |
| 1306 | import threading |
| 1307 | import queue |
| 1308 | op = core.CreateOperator( |
| 1309 | "CreateBlobsQueue", |
| 1310 | [], |
| 1311 | ["queue"], |
| 1312 | capacity=capacity, |
| 1313 | num_blobs=num_blobs, |
| 1314 | device_option=do) |
| 1315 | self.ws.run(op) |
| 1316 | |
| 1317 | xs = [np.random.randn(num_elements, 5).astype(np.float32) |
| 1318 | for _ in range(num_blobs)] |
| 1319 | q = queue.Queue() |
| 1320 | for i in range(num_elements): |
| 1321 | q.put([x[i] for x in xs]) |
| 1322 | |
| 1323 | def enqueue(t): |
| 1324 | while True: |
| 1325 | feed_blobs = ["x_{}_{}".format(i, t) for i in range(num_blobs)] |
| 1326 | op = core.CreateOperator( |
| 1327 | "EnqueueBlobs", |
| 1328 | ["queue"] + feed_blobs, |
| 1329 | feed_blobs, |
| 1330 | device_option=do) |
| 1331 | try: |
| 1332 | elems = q.get_nowait() |
| 1333 | for elem, feed_blob in zip(elems, feed_blobs): |
| 1334 | self.ws.create_blob(feed_blob).feed( |
| 1335 | elem, device_option=do) |
| 1336 | self.ws.run(op) |
| 1337 | except queue.Empty: |
| 1338 | return |
| 1339 | |
| 1340 | # Create all blobs before racing on multiple threads |
| 1341 | # (blob creation is not threadsafe) |
| 1342 | for t in range(num_threads): |
| 1343 | for i in range(num_blobs): |
| 1344 | self.ws.create_blob("x_{}_{}".format(i, t)) |
| 1345 | |
| 1346 | threads = [threading.Thread(target=enqueue, args=(t,)) |
| 1347 | for t in range(num_threads)] |
| 1348 | for thread in threads: |
| 1349 | thread.start() |
| 1350 | |
| 1351 | for n in range(num_elements): |
| 1352 | dequeue_blobs = ["y_{}_{}".format(i, n) for i in range(num_blobs)] |
| 1353 | op = core.CreateOperator( |