Code accompanying Distributed Systems course at KU Leuven
HOST_S  = 'localhost'
PORT_S  = 12345
indirect_communication/event_notification/constants.py
tvcutsem/distributed-systems
from multiprocessing import Process
import zmq
import random
import time
import json
from constants import *

def server():
  context = zmq.Context()         
  socket = context.socket(zmq.PUB)          # create a publisher socket
  socket.bind(f"tcp://*:{PORT_S}")          # bind socket to the address

  time.sleep(5)
  i = 0
  while True:
    data = json.dumps({                     # create JSON object with some fake sensor data
      'seq': i,
      'id': 'sensor123',
      'temp': random.randint(0,50)
    })
    message = "TEMP " + data                # prefix with 'topic' on which subscribers can match
    socket.send(message.encode())           # publish the message
    print(f"server: emit event: '{message}'")
    i += 1
    time.sleep(5)

if __name__ == "__main__":
  server()
indirect_communication/event_notification/publisher.py
tvcutsem/distributed-systems
from multiprocessing import Process
from constants import *
from subscriber import client
from publisher import server

import time

if __name__ == "__main__":
  s  = Process(target=server)
  c1 = Process(target=client)
  c2 = Process(target=client)

  s.start()
  c1.start()
  c2.start()
  
  time.sleep(15)
  c1.terminate()
  c2.terminate()
  s.terminate()

indirect_communication/event_notification/run.py
tvcutsem/distributed-systems
import zmq
import json
from constants import *

def client():
  context = zmq.Context()
  socket = context.socket(zmq.SUB)           # create a subscriber socket
  socket.connect(f"tcp://{HOST_S}:{PORT_S}") # connect to the server
  socket.setsockopt(zmq.SUBSCRIBE, b"TEMP")  # subscribe to TEMP messages (filter on prefix)

  while True:
    msg = socket.recv()                      # receive a message related to subscription
    data = str(msg.decode()).lstrip("TEMP ") # strip the TEMP prefix
    data = json.loads(data)                  # parse the JSON object
    print(f"client: received event: {data}")     
    if (data['temp'] > 30):
      print(f"warning: high temperature measured by sensor {data['id']}")


if __name__ == "__main__":
  client()
indirect_communication/event_notification/subscriber.py
tvcutsem/distributed-systems
BROKER_URL="amqp://guest:guest@localhost:5672/%2f"
QUEUE_NAME='delivery-orders'
indirect_communication/message_queue/constants.py
tvcutsem/distributed-systems
import rabbitpy
import sys
import json
import os

from constants import *
from time import sleep

def consumer(id):
  connection = rabbitpy.Connection(BROKER_URL)
  channel = connection.channel()

  queue = rabbitpy.Queue(channel, QUEUE_NAME)
  queue.declare() # ensure queue exists on the server

  print(f"consumer {id}: connected to queue")

  # while there are messages in the queue, fetch them (using AMQP Basic.Get)
  while len(queue) > 0:
    message = queue.get()
    data = json.loads(message.body.decode())
    print(f"consumer {id}: received order {data}")
    # explicitly acknowledge message receipt after successful processing
    message.ack()
    sleep(1)

if __name__ == "__main__":
  if len(sys.argv) > 1:
     me = sys.argv[1]
  else:
     me = os.getpid()
  consumer(me)
indirect_communication/message_queue/consumer.py
tvcutsem/distributed-systems
import rabbitpy
import time
import random
import json
from constants import *

def producer():
  connection = rabbitpy.Connection(BROKER_URL) # Connect to RabbitMQ server
  channel = connection.channel()     # Create new channel on the connection

  queue = rabbitpy.Queue(channel, QUEUE_NAME) # Create a work queue
  queue.declare() # make sure it exists on the server

  time.sleep(5)
  for i in range(100):
    data = {'order_id': i, 'total_value': random.randint(0,50)}
    message = rabbitpy.Message(channel, json.dumps(data)) # serialize data to string
    # Publish the message directly to the queue
    message.publish('', QUEUE_NAME)
    print(f"producer: submitted order '{data}'")
    time.sleep(5)

if __name__ == "__main__":
	producer()
indirect_communication/message_queue/producer.py
tvcutsem/distributed-systems
from multiprocessing import Process
from producer import producer
from consumer import consumer
from time import sleep

# Make sure that a RabbitMQ server is running, or otherwise
# this program will fail (see README)

if __name__ == "__main__":
	p = Process(target=producer)
	c = Process(target=consumer, args=('1',))
	p.start()

	sleep(6)
	c.start()
	
	p.join()
	c.join()
	

indirect_communication/message_queue/run.py
tvcutsem/distributed-systems
SRC1  = "127.0.0.1"
SRC2  = "127.0.0.1"
PORT1 = "50001"
PORT2 = "50002"
HOST_S = "127.0.0.1" # Note: ZMQ PUSH does not work with string 'localhost'
PORT_S = "12345"
indirect_communication/pipeline/constants.py
tvcutsem/distributed-systems
import zmq
import json
import sys
import os

from constants import *

def consumer(pid):
  context = zmq.Context()
  s = context.socket(zmq.PULL)       # create a pull socket
  s.connect(f"tcp://{SRC1}:{PORT1}") # connect to task source 1
  # s.connect(f"tcp://{SRC2}:{PORT2}") # connect to task source 2

  while True:
    print(f"consumer {pid}: waiting for orders")
    order = json.loads(s.recv().decode())  # receive msg from any task source
    print(f"consumer {pid}: received order {order}")

if __name__ == "__main__":
  pid = os.getpid()
  consumer(pid)
indirect_communication/pipeline/consumer.py
tvcutsem/distributed-systems
import zmq
import time
import json
import random
import sys

from constants import *

def producer(port):
    context = zmq.Context()              
    socket  = context.socket(zmq.PUSH)      # create a push socket
    socket.bind(f"tcp://{HOST_S}:{port}")   # bind socket to address
    
    time.sleep(2)
    for i in range(10):                     # generate 10 orders
      order = json.dumps({
          'src': port,
	        'order_id': i,
          'total_value': random.randint(0,50)
	    })
      print(f"producer: submitting order '{order}'")
      socket.send(order.encode())
      time.sleep(5)

if __name__ == "__main__":
    if len(sys.argv) > 1:
      me = str(sys.argv[1])
    else:
      me = PORT1
    producer(me)

indirect_communication/pipeline/producer.py
tvcutsem/distributed-systems
from multiprocessing import Process
from constants import *
import zmq, time, pickle, random

NWORKERS = 10

def producer():
  context = zmq.Context()              
  socket  = context.socket(zmq.PUSH)      # create a push socket
  socket.bind(f"tcp://{HOST_S}:{PORT_S}") # bind socket to address
  
  while True:
    workload = random.randint(1, 30)     # compute workload
    print("Producer: created work", format(workload,'03d'))
    socket.send(pickle.dumps(workload))   # send workload to worker
    time.sleep(workload/NWORKERS)         # balance production by waiting 

def worker(id):
  context = zmq.Context()
  socket  = context.socket(zmq.PULL)         # create a pull socket
  socket.connect(f"tcp://{HOST_S}:{PORT_S}") # connect to the producer
  workerID = format(id,'03d')

  while True:
    print(f"Worker {workerID}: asking for work")
    work = pickle.loads(socket.recv())     # receive work from a source
    print(f"Worker {workerID}: got work {format(work,'03d')}")
    time.sleep(work)                       # pretend to work
    
if __name__ == "__main__":
  prod = Process(target=producer)
  workers = [Process(target=worker,args=(i+1,)) for i in range(NWORKERS)]

  for i in range(NWORKERS):
    workers[i].start()
  prod.start()

  time.sleep(30)
  for i in range(NWORKERS):
    workers[i].terminate()
  prod.terminate()


indirect_communication/pipeline/run.py
tvcutsem/distributed-systems
BROKER_URL="amqp://guest:guest@localhost:5672/%2f"
EXCHANGE_ID="sensor_data"
indirect_communication/publish_subscribe/constants.py
tvcutsem/distributed-systems
import rabbitpy
import time
import random
import json
from constants import *

def publisher():
  connection = rabbitpy.Connection(BROKER_URL) # Connect to RabbitMQ server
  channel = connection.channel()     # Create new channel on the connection

  # Create a "topic" exchange to post sensor data
  exchange = rabbitpy.Exchange(channel, EXCHANGE_ID, exchange_type='topic')
  exchange.declare() # ensure exchange exists on the server
  # can remove using exchange.delete()

  time.sleep(5)
  for i in range(100):
    data = {'seq': i, 'id': 'sensor123', 'temp': random.randint(0,50)}
    message = rabbitpy.Message(channel, json.dumps(data)) # serialize data to string
    # Publish the message using a "routing key"
    print(f"publisher: emitting event '{data}'")
    message.publish(exchange, 'temp.sensor123')
    time.sleep(5)

if __name__ == "__main__":
	publisher()
indirect_communication/publish_subscribe/publisher.py
tvcutsem/distributed-systems
from multiprocessing import Process
from publisher import publisher
from subscriber import subscriber
from time import sleep

# Make sure that a RabbitMQ server is running, or otherwise
# this program will fail (see README)

if __name__ == "__main__":
	p = Process(target=publisher)
	c1 = Process(target=subscriber, args=('1',))
	c2 = Process(target=subscriber, args=('2',))
	p.start()

	sleep(10)
	c1.start()
	c2.start()
	
	p.join()
	c1.join()
	c2.join()	

indirect_communication/publish_subscribe/run.py
tvcutsem/distributed-systems
import rabbitpy
import sys
import json
from constants import *
import os

def subscriber(id):
  with rabbitpy.Connection(BROKER_URL) as conn:
      with conn.channel() as channel:
          queue = rabbitpy.Queue(channel, f'subscriber-{id}')
          queue.declare() # ensure queue exists on the server
          queue.bind(EXCHANGE_ID, 'temp.*') # Bind queue to exchange via routing key

          print(f"subscriber {id}: connected to queue")

          try:
              # Consume messages in queue (blocking if none are available)
              for message in queue:
                data = json.loads(message.body.decode())
                print(f"subscriber {id}: got message {data}")
                if (data['temp'] > 30):
                    print(f"warning: high temperature measured by sensor {data['id']}")
                # explicitly acknowledge message receipt after successful processing
                message.ack()

          # Exit on CTRL-C
          except KeyboardInterrupt:
              print(f'subscriber {id}: stopped')


if __name__ == "__main__":
  if len(sys.argv) > 1:
     me = sys.argv[1]
  else:
     me = os.getpid()
  subscriber(me)
indirect_communication/publish_subscribe/subscriber.py
tvcutsem/distributed-systems
from multiprocessing import Process, Pipe
import os

def server(conn):
	print(f"Hello from server process {os.getpid()}")
	data = conn.recv()
	print(f"server received data: {data}")
	conn.close()

def client(conn):
	print(f"Hello from client process {os.getpid()}")
	values = [42, "hello world", {"key": "value"}]
	conn.send(values)
	print(f"client sent data {values}")
	conn.close()

if __name__ == "__main__":
    # construct a pipe to connect two processes
	(p1_conn, p2_conn) = Pipe()
	p1 = Process(target=server, args=(p1_conn,))
	p2 = Process(target=client, args=(p2_conn,))
	# run both processes in parallel
	p1.start()
	p2.start()
	# wait for both processes to finish
	p1.join()
	p2.join()

  

  

indirect_communication/python_basics/ipc/run.py
tvcutsem/distributed-systems
from multiprocessing import Process
import os

def info():
    print('module name:', __name__)
    print('parent process:', os.getppid())
    print('process id:', os.getpid())

def hello(name):
    print('hello ', name)
    info()

if __name__ == '__main__':
    info()
    p = Process(target=hello, args=('world',))
    p.start()
    p.join()
indirect_communication/python_basics/ipc/spawn.py
tvcutsem/distributed-systems
from socket  import *
from constants import *

class Client:
  def run(self):
    s = socket(AF_INET, SOCK_STREAM)
    print(f"client: connecting to server at {HOST}:{PORT}")
    s.connect((HOST, PORT)) # connect to server (block until accepted)
    s.send(b"Hello, world") # send same data
    data = s.recv(1024)     # receive the response
    # print what you received
    print(f"client: received data from server: {data}")
    s.send(b"")             # tell the server to close
    s.close()               # close the connection

if __name__ == "__main__":
  c = Client()
  c.run()
indirect_communication/python_basics/sockets/client.py
tvcutsem/distributed-systems
import multiprocessing
from server import *
from client import *

def server():
	s = Server()
	s.run()

def client():
	c = Client()
	c.run()

if __name__ == "__main__":
	s = multiprocessing.Process(target=server)
	c = multiprocessing.Process(target=client)

	s.start()
	c.start()
	s.join()
	c.join()

  

  

indirect_communication/python_basics/sockets/run.py
tvcutsem/distributed-systems
from socket  import *
from constants import *

class Server:
  def run(self):
    s = socket(AF_INET, SOCK_STREAM) 
    s.bind((HOST, PORT))
    s.listen(2)
    print(f"server: accepting client connections at {HOST}:{PORT}")
    (conn, addr) = s.accept()  # returns new socket and addr. client 
    while True:                # forever
      data = conn.recv(1024)   # receive data from client
      if not data: break       # stop if client stopped
      print(f"server: received data from client: {data}")
      conn.send(data+b"!")     # return sent data plus an "!"
    print(f"server: shutting down")
    conn.close()               # close the connection

if __name__ == "__main__":
  s = Server()
  s.run()
indirect_communication/python_basics/sockets/server.py
tvcutsem/distributed-systems
import zmq
from constants import *

def client():
  context = zmq.Context()
  socket  = context.socket(zmq.REQ)       # create request socket
  socket.connect(f"tcp://{HOST_S}:{PORT_S}") # block until connected

  socket.send(b"Hello world")             # send message
  message = socket.recv()                 # block until response
  socket.send(b"STOP")                    # tell server to stop
  print(message.decode())                 # print result


if __name__ == "__main__":
  client()
indirect_communication/request_reply/client.py
tvcutsem/distributed-systems
HOST_S  = 'localhost'
PORT_S  = 12345
indirect_communication/request_reply/constants.py
tvcutsem/distributed-systems
# helper file to start client and server in one go as subprocesses

from multiprocessing import Process
from time import sleep
from client import client
from server import server

if __name__ == "__main__":
  s = Process(target=server)
  c = Process(target=client)
  s.start()
  sleep(2)
  c.start()
  c.join()
  s.join()

indirect_communication/request_reply/run.py
tvcutsem/distributed-systems
import zmq
from constants import *

def server():
  context = zmq.Context()
  socket  = context.socket(zmq.REP)       # create reply socket
  socket.bind(f"tcp://*:{PORT_S}")        # bind socket to address

  while True:
    message = socket.recv()               # wait for incoming message
    if not "STOP" in str(message):        # if not to stop...
      print(f"received: {message}")
      reply = str(message.decode())+'!'   # append "!" to message
      socket.send(reply.encode())         # send it away (encoded)
    else:                         
      break                               # break out of loop and end

if __name__ == "__main__":
  server()
indirect_communication/request_reply/server.py
tvcutsem/distributed-systems
OK = 0
ERR = 1
WRITE = 2
READ = 3
TAKE = 4

COMMANDS = ["OK", "ERR", "WRITE", "READ", "TAKE"]
indirect_communication/tuple_spaces/constants.py
tvcutsem/distributed-systems
import time
import random
from multiprocessing import Process
from ts_server   import TSServer
from ts_client   import TSClient

HOST = "localhost"
PORT = 5678

# define the processes

def server_process():
  server = TSServer(PORT)
  server.run()

def publisher_process():
  ts = TSClient(HOST, PORT)
  time.sleep(5)
  for i in range(100):
    # ('temp', sequence number, sensor id, temperature value)
    event = ('temp', i, 'sensor123', random.randint(0,50))
    print(f"publisher: emit event: '{event}'")
    ts.write(event)
    # clean up old events...
    old_event = ('temp', i-3, 'sensor123', None)
    ts.take(old_event, probe=True) # if the old_event does not exist, don't block
    time.sleep(5)

def subscriber(id):
  ts = TSClient(HOST, PORT)
  i = 0
  while True:
    if i == 0:
      # read first update
      (_, seq, _, temp) = ts.read(('temp', None, 'sensor123', None))
      i = seq
      print(f"subscriber {id}: read temperature update {seq}: {temp}")
    else:
      # read next update
      (_, seq, _, temp) = ts.read(('temp', i, 'sensor123', None))
      print(f"subscriber {id}: read temperature update {seq}: {temp}")
    i = i + 1
  
if __name__ == "__main__":
  processes = [
    Process(target=server_process),
    Process(target=publisher_process),
    Process(target=subscriber, args=("A",)),
    Process(target=subscriber, args=("B",))
  ]
  for p in processes:
    p.start()


indirect_communication/tuple_spaces/event_example.py
tvcutsem/distributed-systems
import pickle
import zmq
from constants import *
context = zmq.Context()

class TSClient:
  def __init__(self, host, port):
    self.socket = context.socket(zmq.REQ)        # create request socket
    self.socket.connect(f"tcp://{host}:{port}")  # block until connected

  def __sendrecv(self, message):            # this is a private method
    self.socket.send(pickle.dumps(message)) # serialize and send
    message = self.socket.recv()            # block until response
    return pickle.loads(message)            # deserialize response

  def write(self, tuple):
    """write a tuple. Never blocks."""
    reply = self.__sendrecv([WRITE, tuple])
    if reply[0] == OK:                      # check status code
      return None
    else: # reply[0] == ERR
      raise ValueError(reply[1])
  
  def take(self, pattern, probe=False):
    """remove matching tuple, blocking if none match and probe is False"""
    status, reply = self.__sendrecv([TAKE, pattern, probe])
    if status == OK:
      return reply
    else: # status == ERR
      raise ValueError(reply)
    
  def read(self, pattern, probe=False):
    """read matching tuple, blocking if none match and probe is False"""
    status, reply = self.__sendrecv([READ, pattern, probe])
    if status == OK:
      return reply
    else: # status == ERR
      raise ValueError(reply)

indirect_communication/tuple_spaces/ts_client.py
tvcutsem/distributed-systems
import pickle
import zmq
from constants import *

context = zmq.Context()

# elementwise comparison using '=='
# None in the pattern matches everything
def match(tuple, pattern):
  if len(tuple) == len(pattern):
    for i in range(len(tuple)):
      if pattern[i] != None and tuple[i] != pattern[i]:
        return False
    return True
  else:
    return False

class TSServer:
  def __init__(self, port=5678):
    self.socket = context.socket(zmq.ROUTER)
    self.socket.bind(f"tcp://*:{port}")   # bind socket to address
    self.tuplespace = []                  # a Python list of Pyton tuples
    self.waitingTakers = []
    self.waitingReaders = []
    print(f"tsserver: listening on port {port}")

  def __write(self, tuple):
    # check if there are outstanding READ requests that match
    matchingReaders = [(s,patt) for (s,patt) in self.waitingReaders if match(tuple, patt)]
    # reply to the pending requests (unblocking the processes)
    for pair in matchingReaders:
        sender, _ = pair
        self.__reply(sender, [OK, tuple])
        # remove the pair from the wait list
        self.waitingReaders.remove(pair)

    # check if there are outstanding TAKE requests that match
    # only the FIRST one can fire...
    matchingTaker = next(filter(lambda pair: match(tuple, pair[1]), self.waitingTakers), None)
    if matchingTaker is not None:
      (sender, patt) = matchingTaker
      self.__reply(sender, [OK, tuple])
      self.waitingTakers.remove(matchingTaker)
    else:
      # no waiting TAKE requests match, add the tuple
      self.tuplespace.append(tuple)

  def __take(self, pattern):
    matchingTuple = next(filter(lambda tuple: match(tuple, pattern), self.tuplespace), None)
    if matchingTuple is not None:
      # remove tuple and then return it
      self.tuplespace.remove(matchingTuple)  
      return matchingTuple
    else:
      # no matching tuple found
      return None
  
  def __read(self, pattern):
    for tuple in self.tuplespace:
      if match(tuple, pattern):
        # return tuple without removing it
        return tuple
    # no matching tuples found
    return None

  def __reply(self, sender, result):
    # send a response in ZMQ's RESP format
    # frames: [receptionist id, empty frame, data frame]
    self.socket.send_multipart([
      sender,
      bytes(),
      bytes(pickle.dumps(result))])

  def run(self):
    while True:
      message = self.socket.recv_multipart()        # wait for incoming message
      # as the other peer is a REQ socket, the ROUTER receives a 3-part message:
      # [ sender identity frame, empty frame, data frame ]
      sender = message[0]
      request = pickle.loads(message[2])     # parse message into Python object
      # request = [COMMAND, tuple, probe?]
      command = request[0]

      print(f"tsserver: received command '{COMMANDS[command]}'")

      if command == WRITE:
        tuple = request[1]
        self.__write(tuple)
        self.__reply(sender, [OK])
      
      elif command == TAKE:
        pattern = request[1]
        probe = request[2]
        result = self.__take(pattern)
        if result is not None:
          self.__reply(sender, [OK, result])
        else:
          if probe:
            # just probing, respond immediately
            self.__reply(sender, [OK, None])
          else:
            # delay reply until tuple exists
            self.waitingTakers.append((sender, pattern))
      
      elif command == READ:
        pattern = request[1]
        probe = request[2]
        result = self.__read(pattern)
        if result is not None:
          self.__reply(sender, [OK, result])
        else:
          if probe:
            # just probing, respond immediately
            self.__reply(sender, [OK, None])
          else:
            # delay reply until tuple exists
            self.waitingReaders.append((sender, pattern))
      
      else:
        self.__reply(sender, [ERR, "invalid command"])


if __name__ == "__main__":
  s = TSServer()
  s.run()
indirect_communication/tuple_spaces/ts_server.py
tvcutsem/distributed-systems
from multiprocessing import Process
from ts_server   import TSServer
from ts_client   import TSClient

HOST = "localhost"
PORT = 5678

# helper functions

def factorial(n):
  res = 1
  while n > 0:
    res = res * n
    n = n - 1
  return res

def fibonacci(n):
    if n in {0, 1}:
        return n
    prev, next = 0, 1
    for _ in range(2, n + 1):
        prev, next = next, prev + next
    return next

# define the processes

def server_process():
  server = TSServer(PORT)
  server.run()

def master_process():
  ts = TSClient(HOST, PORT)
  ts.write(("fac", 5))
  ts.write(("fib", 3))
  fac_res, fac = ts.take(("fac_res", None))
  print(f"master: got fac result: {fac}")
  fib_res, fib = ts.take(("fib_res", None))
  print(f"master: got fib result: {fib}")

def fac_worker():
  ts = TSClient(HOST, PORT)
  fac, n = ts.take(("fac", None))
  print(f"fac_worker: got work: fac({n})")
  result = factorial(n)
  ts.write(("fac_res", result))

def fib_worker():
  ts = TSClient(HOST, PORT)
  fac, n = ts.take(("fib", None))
  print(f"fib_worker: got work: fib({n})")
  result = fibonacci(n)
  ts.write(("fib_res", result))
  
if __name__ == "__main__":
  processes = [
    Process(target=server_process),
    Process(target=master_process),
    Process(target=fac_worker),
    Process(target=fib_worker)
  ]
  for p in processes:
    p.start()


indirect_communication/tuple_spaces/worker_example.py
tvcutsem/distributed-systems