Does the per-thread getting of a value, running the sub-function, and marking a completion.
(logger, tnum, q, port, command, zones_struct, zgrab_collection)
| 360 | |
| 361 | |
| 362 | def process_data(logger, tnum, q, port, command, zones_struct, zgrab_collection): |
| 363 | """ |
| 364 | Does the per-thread getting of a value, running the sub-function, and marking a completion. |
| 365 | """ |
| 366 | while not global_exit_flag: |
| 367 | global_queue_lock.acquire() |
| 368 | if not global_work_queue.empty(): |
| 369 | data = [] |
| 370 | i = 0 |
| 371 | while i < 50: |
| 372 | data.append(q.get()) |
| 373 | i = i + 1 |
| 374 | if global_work_queue.empty(): |
| 375 | break |
| 376 | global_queue_lock.release() |
| 377 | logger.debug("Thread %s processing %s" % (str(tnum), data)) |
| 378 | try: |
| 379 | process_thread( |
| 380 | logger, data, port, command, zones_struct, zgrab_collection, tnum |
| 381 | ) |
| 382 | except Exception as ex: |
| 383 | logger.warning("Thread error processing: " + str(data)) |
| 384 | logger.warning(str(ex)) |
| 385 | for _ in range(0, i): |
| 386 | q.task_done() |
| 387 | else: |
| 388 | global_queue_lock.release() |
| 389 | time.sleep(1) |
| 390 | |
| 391 | |
| 392 | class ZgrabThread(threading.Thread): |