StreamScope/main.py at main · mozenjii/StreamScope · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
import multiprocessing as mp
from plugins.inputs import CSVProducer
from plugins.outputs import DashboardGUI
from core.aggregator import aggregator_process
from core.worker import worker_process
from core.configuration import read_config
from core.telemetry import PipelineTelemetry


def main():

    config = read_config()
    queue_max_size = config["pipeline_dynamics"]["stream_queue_max_size"]
    ctx = mp.get_context("spawn")

    raw_queue = ctx.Queue(maxsize=queue_max_size)
    verified_queue = ctx.Queue(maxsize=queue_max_size)
    processed_queue = ctx.Queue(maxsize=queue_max_size)

    read_count = ctx.Value('i', 0)
    verified_count = ctx.Value('i', 0)
    dropped_count = ctx.Value('i', 0)
    stop_packet = {"_control":"STOP"}

    producer = CSVProducer(config, raw_queue, stop_packet, read_count)

    producer_process = ctx.Process(target=producer.run,name="ProducerProcess")

    producer_process.start()


    worker_processes = []
    for worker_id in range(config["pipeline_dynamics"]["core_parallelism"]):
        worker = ctx.Process(target=worker_process, args=(config, raw_queue, verified_queue, verified_count, dropped_count, stop_packet), name=f"WorkerProcess-{worker_id}")
        worker_processes.append(worker)
        worker.start()

    aggregator = ctx.Process(
        target=aggregator_process,
        args=(config, verified_queue, processed_queue, stop_packet),
        name="AggregatorProcess",
    )
    aggregator.start()

    dashboard = DashboardGUI(config, processed_queue, stop_packet)
    telemetry = PipelineTelemetry(
        config,
        raw_queue,
        verified_queue,
        processed_queue,
        read_count=read_count,
        verified_count=verified_count,
        dropped_count=dropped_count,
    )
    telemetry.attach(dashboard)
    telemetry.poll()
    telemetry.start()

    try:
        dashboard.run()
    finally:
        telemetry.stop()
        telemetry.detach(dashboard)

    producer_process.join()
    for worker in worker_processes:
        worker.join()
    aggregator.join()

    print("Rows read:", read_count.value)
    print("Verified packets:", verified_count.value)
    print("Dropped packets:", dropped_count.value)
    print("Raw queue size:", raw_queue.qsize())
    print("Verified queue size:", verified_queue.qsize())
    print("Processed queue size:", processed_queue.qsize())


if __name__ == "__main__":
    main()