-
Notifications
You must be signed in to change notification settings - Fork 1
/
worker.py
60 lines (51 loc) · 1.6 KB
/
worker.py
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
import socket
import logging
import sys
import json
import threading
from pprint import pformat
import time
from dataclasses import asdict
from job_utils.task import Task
from job_utils.task import WorkerMessage
logging.basicConfig(
format="%(asctime)s: %(message)s", level=logging.INFO, datefmt="%Y-%m-%dT%H:%M:%S%z"
)
port = int(sys.argv[1])
w_id = int(sys.argv[2])
responsePort = 5001
responseHost = "localhost"
logging.info("Starting worker %d on port %d", w_id, port)
master = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
host = "127.0.0.1"
master.bind((host, port))
def waste_time(task: Task):
time.sleep(task.duration)
logging.info("Completed task %s on worker %d", task.task_id, w_id)
msg = WorkerMessage(
addr=[host, port],
message="Completed task {} on worker {}".format(task.task_id, w_id),
task_id=task.task_id,
w_id=w_id,
job_id=task.job_id,
)
msg = json.dumps(asdict(msg))
master_send = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
master_send.connect((responseHost, responsePort))
master_send.sendall(msg.encode())
master_send.close()
master.listen()
while True:
conn, addr = master.accept()
with conn:
logging.info("Connected by %s", addr)
data = conn.recv(1024)
if not data:
continue
data = json.loads(data)
task = Task(**data)
logging.info("Got task: %s", pformat(task))
# task_id, duration
logging.info("starting task %s on worker %d", task, w_id)
thread = threading.Thread(target=waste_time, args=(task,))
thread.start()