MPCS 52040 Distributed Systems Midterm

5 Characteristics of Distributed Systems

1) more than one computing element 2) independent computing elements that operate concurrently 3) no global clock (each node has own notion of time, may be different, unpredictable communication latency) 4) independent failures 5) single coherent system (to user)

Middleware

distributed OS, manages the distributed system and provides services to apps hosted on distributed system

Design Goals of Distributed Systems

1) Share: make remote resources available (share) 2) Transparent: hide that processes & resources are physically distributed across multiple computers 3) Open: interoperability; be open such that compoenents can easily be used by others (well-defined interfaces, portable, extensible) 4) Scalable: scales to increased number of users, resources, locations 5) Dependable: able to deliver reliable & correct processing (available, reliable, fails safely, repairable) 6) Secure: resources, computations, users are protected (CIAA triad)

High Performance Distributed Computing System

combines parallel supercomputing power with a network of multiple independent computers (nodes) that work together to solve massive, complex problems e.g. clusters, grids, clouds

Cluster computing

group of high-end systems connected through LAN

• homogenous: same OS, near identical hardware

• single managine node

Grid computing

pulling all clusters together into 1 distributed system. use virtual organizations: grouping of users that will allow for authorization on resource allocation. many nodes everywehere, hetereogenous, dispered on WAN

Cloud computing

the delivery of shared computing resources (including data storage, servers, databases, networking, and software) over the internet on a pay-as-you-go basis, replacing the need to buy and maintain physical hardware i.e. outsource entire infrastructure that is needed for application (rent out infrastructure like datacenters, computation, software)

Distributed Information Systems

a collection of independent computers, software applications, and databases spread across a network that communicate and cooperate to appear as a single, unified system to the user - distributed transactions, enterprise application integration

Distributed System for Pervasive Computation

a network of independent, often embedded or mobile devices that collaborate seamlessly in the background to provide computing services anywhere and anytime e.g. mobile computing, edge devices, sensor networks

Sensor Networks

a group of interconnected devices that collect, process, and transmit data about physical or environmental conditions to a central location. contrained - wireless communication, resource constrained (battery powered), scope-based communication

Edge computing

a distributed computing model that brings data storage and processing physically closer to the source--such as local devices, sensors, or edge servers--rather than relying solely on a distant centralized cloud or data center e.g. IoT, sensor networks, for private data

Layered architectures

components are organized into layers. higher layers make calls to lower layers, responses from lower layers are returned to higher layers. e.g. most applications (python program -> python libraries -> OS calls), network communication architectures

Common layering approach

Application interface layer: user interface Processing layer: logic and function of application (database queries) Data layer: state manipulated by application components (database)

Layered architectures pros and cons

PROS: simple implementation, widely used, intuitive CONS: strong dependency between layers --> bottlenecks, modularity, layer changes

Service-oriented architectures

applications built from discrete, self-contained, and reusable software components called services that communicate over a network. similar to object-based architectures (components are objects, connected through procedural calls)

How is information exchanged across sevice-oriented architecture (SOA)?

Simple object access protocol (SOAP) is strict protocol for echanging info across SOA

Microservice

small independent services responsible for a specific business function (i.e. one service does on thing) services communicate over lightweight channels (REST, gRPC, message queues) each service can be scaled, deployed, maintained independently. good for fault tolerance and stability

REST (Representational State Transfer)

A distributed system as a collection of resources invidividually managed by components. Resources may be added, removed, retrieved, modified by remote applications. Unlike SOAP, is a flexible architecture (not a strict protocol)

Characteristics of REST

1) resources identified through single naming scheme 2) all services offer same interface 3) messages sent to or from a service are fully self-described 4) after executing an operation at a service, that component forgets everything about the caller

REST on HTTP

POST: create new resource GET: retrieve state of resource DELETE: delete a resource PUT: modify resource by transferring new state

REST example

Amazon S3 (simple storage service) provides object storage for
1) objects - bundle of bytes representing something
2) buckets - namespace for organizaing objects

simple operations to GET, POST, PUT, DELETE

Publish-subscribe architectures

decoupled dependencies between coordination and processing - message publishers and message subscribers use an intemediary component instead of communicating directly (e.g. message broker, event bus)

Referentially coupled, temporally coupled

Direct - explicit reference in communication. process can communicate if know name of other process. both processes are online at same time

Referentially coupled, temporally decoupled

Mailbox - processes not executing at same time but still putting message in known mailbox of other process

Referentially decoupled, temporally coupled

Event-based - processes don't know each other explicitly, but publish notification describing occurence of some event. other processes can choose to subscribe to specific type of event to receive notifications about that event

Referentially decoupled, temporally decoupled

Shared data space - processes communicate via tuples (data records with number of fields). one process puts in tuple space, another process can search for it using patterns

Process

a program in execution (instance of a computer program that is being executed)

What is in a process

• machine code that's being run (code and instructions)
• memory allocation (data, call stack, heap)
• resources (files, devices)
• processor context (registers, memory addresses)

stored in process control block (PCB)

start --> ready

Process pros and cons

PRO: OS ensures processes are independent (H/W support), provide strong isolation CON: creating a process == OS must create new address space, context switching b/t processes requires saving and reloading context, heavyweight

Processor

physicial hardware that executed instructions

virtual processor (vCPU)

logical CPU presented to a guest OS by the hypervisor. scheduled onto physical cores, so each VM appears to have its own processor. can be oversubscribed (more vCPUs than cores): better utilization, but vCPUs may wait under load

virtual machine

emulator of a computer system that is an efficient, isolated duplicate allows users to run multiple OS's on single machine, allows apps to run separate from your hardware, fast deployment (app portability)

Pros and cons of using virtual machine

PROS: strong isolation - each VM runs own sandboxed environment so crashes isolated to single VM platform flexibility - can run multiple OS's on single machine

CONS: high resource consumption - consumes dedicated RAM, CPU, storage per VM performance overhead - hypervisor and guest OS consume extra CPU, RAM, and storage which makes VM slower than bare apps large size compared to containers - GBs vs. MBs startup time is longer than containers - minutes vs. seconds/milliseconds

hypervisor

a program used to run and manage one or more virtual machines on a computer

• can reallocate resources (physical hardware like CPU, RAM, storage) between VMs

hardware assisted VM

provide architectural support for hypervisors to run VMs faster and with less overhead (allow guest OSes to be run in isolation)

process context

context: program counter, registers, memory maps, open files (stored in PCB)

context switching

save current state of process/thread so can be later restored when execution resules, load the next process/thread's state process -> OS switches context (interrupts kernel) threads -> share address space so independent of OS

overhead: no useful work during the switch, plus cache/TLB disruption

Why do we need multiple processes?

whenever a blocking system call is executed the entire process blocks -- apps need to do a lot of things concurrently multiprocessor/multicore machines allow independent functions to be executed concurrently (functions can run in parallel --> overall execution faster)

Thread (properties)

1) executes its own piece of code
2) independently of other threads
3) not much concurrency transparency (user sees the complexity, needs to handle isolation)
4) maintains only minimal context that allows the CPU to be shared (only processor context and minimal info for thread management)
5) threads can access shared data in the process

tradeoff between process and threads

performance vs. isolation

• process has strong isolation

• but threads have better performance (faster to create & context switch)

• creating/destroying threads cheaper than processes

Threads in user space

many to one model: many threads mapped to one kernel process

Threads in user space pros and cons

PROS

• cheap: all admin kept in user address space (cost is just of allocating memory for thread's stack)

• efficient: all operations handled within single process, all services provided by the kernel are done on behald of that process

• context switching: can be done in few instructions (only CPU register vals need to be stored/reloaded) CONS

• threads mapped to single schedulable entity at OS level

• invocation of blocking sys call would block whole process the threads belongs to

Threads in kernel space ( ? to ?)

One-to-one model: every thread is a schedulable entity kernel implements the thread package so all operations return as system calls

Threads in kernel space pros and cons

PROS

• operations that block a thread are no longer a problem - if thread A blocks, the kernel simply pauses thread A and immediately schedules thread B (from the same process) to keep running.

• handling external events is simple - kernel can schedule the thread waiting for the specific event

CONS

• each thread operation (create, delete, sync) has to be carried out by kernel using a system call

• switching contexts now as expensive as switching process context

Multi-threaded clients

hiding network latencies by having clients do several calls at the same time, each one by a different thread. wait until all results are returned e.g. web browser scans HTML page to find elements that need to be fetched, each file is fetched by separate thread, each doing a (blocking) HTTP request, as file comes in the browser displays them

multiple request-response calls to other machine (RPC) if calls are to different servers, may have a linear speed-up

Multi-threaded servers properties

hide network latencies by reacting to next request while previous one is being replied to to improve performance

• scales better: as opposed to single-threaded server, simple scale-up to multiprocessor system

• better structure: most servers have high I/O demands, so using simple blocking calls simplifies the overall structure

• simpler: multithreaded programs tend to be smaller and easier to understand b/c of simplified flow of control

Virtualization

enables you to run a virtual computer system using an abstraction layer above the hardware

• extend/replace an existing interface to mimic the behavior of another system

Why is virtualization important?

1) hardware changes faster than software 2) ease of portability and code migration 3) isolation of failing or attacked components 4) provides elastic computing capacity in the cloud and ability to manage many heterogeneous servers

Why do we say that virtualization can occur at many levels?

Applications use library functions as an abstraction layer above library Library uses system calls as an abstraction layer above OS OS uses privileged instructions and general instructions as layer above hardware

containers

lightweight, isolated environment that packages an application and its dependencies, shares the host's OS kernel

e.g. Docker

Properties of containers

Namespaces (isolation): each container gets own view of PIDs, network mounts, hostname, users Cgroups (limits): cap and account for CPU, memory, disk I/O, process couns so one container can't starve others Shared host kernel: no guest OS --> much lower overhead than traditional VMs

Virtual Machines vs Containers

Architecture: VM emulates hardware with a hypervisor which runs a full guest OS, container shares the host OS kernel using namespaces and cgroups

Size: VM much bigger (GB), container (MB)

Startup time: VM minutes, container seconds/ms

Resource usage: High for VM (consumes dedicated RAM, CPU, storage per VM), Low for containers (lightweight, runs only the app and its dependencies)

Isolation: VM strong (hardware-level isolation), container moderate (process-level isolation, shares host kernel)

Docker

enterprise containers developed for linux uses cgroups to limit resource usage (CPU/mem) kernel namesoaces limit app view of the operating environment (process trees, networks, file systems)

Why was virtualization important for cloud computing?

Provieded framework of infrastructure as service -->

• cloud providers rent out VM
• physical hardware separated from the software that runs on it
• allows cloud providers to share physical hardware safely with multiple customers while accounting for isolation, resource sizing

Thick client (with examples)

most compution and application logic run locally application state stored mainly on client reduced dependence on continuous connectivity

e.g. microsoft word, video games, photoshop

Thin client (with examples)

Most compution and application logic run on servers application state managed centrally requires continuous network communication

e.g. google docs, dropbox, youtube

Transparency

ability of the system to hide its complex multi-node architecture from users - appear like one single, unified computer user can "see through" complexity i.e. it is transparent to user so interacts with data as if it is one system

How might a client provide transparency on the client-side?

1) access transparency: client stubs for RPC, so client can use same commands to access remote resources 2) location/migration transparency: client tracks actual location 3) replication transparency: if multiple servers, client will handle replicated invocations to each server 4) failure transparency: client must support if masking server/communication failures

Server

a process implementing a specific service on behalf of a collection of clients

Iterative server

server handles one request before attending to next request

Concurrent server

uses a dispatcher to pick up incoming request, passes it on to separate process or thread

How to contact a server?

server listens on a specific port. clients will know the port based on well-known ports or discovering port somehow

Well known ports

SSH == TCP 22 HTTP == TCP 80 HTTPS == TCP 443 SMTP == TCP 25 DNS == TCP/UDP 53

how are servers discovered?

client needs to first ask for the port server registers service endpoint to a daemon process superserver manages creation of server on port within predefined range

interrupting a server after service request already accepted?

1) put computer on airpane mode, server eventually could time out 2) use separate port for urgent data, then mistake request is put on hold (OS needs to support priority-based scheduling) 3) use transport layer - TCP allows urgent messages in same connection

stateless server (and example)

does not keep information on the state of client can change state without informing clients in practice, most servers store some info on client state, but if state is lost service wont be disrupted

e.g. HTTP and REST APIs - servers read payload from client request, perform action, return response, immediately forget interaction

Pros and cons of stateless server

PROS

• independence: of clients and servers
• inconsistency: if client/server crashes no need to sync

CONS

• performance: server cannot anticipate client behavior by precomputing

stateful servers (and example)

server maintains client-specific information across requests until explicitly deleted e.g. file server tracks open files for efficient pre-fetching, knows what data clients have cached for faster access

stateful server pros and cons

PRO

• better performance via caching and prefetching (can anticipate client behavior)

CON

• lower reliability if state data lost or corrupted

server clusters

in practice, many interconnected servers work together to handle client requests - share workloads and increase perfomance

dockerfile for building Ubuntu container image and run test.sh

FROM ubuntu:latest RUN apt-get update -y RUN chmod +x /test.sh CMD ["/test.sh"]

In HW 1, the server maintains the game board while the clients send moves. Why is this a client-server architecture rather than a peer-to-peer architecture. What are pros and cons of this design?

This is client server because the server is the centralized authority responsible for storing and validating the game state. clients send move requests but do not independently maintain the game state. peer-to-peer would require coordination between peers to agree on moves and maintain consistency together.

PROS: consistent game state and centralized rule enforcement CONS: single point of failure, potential performance bottleneck

In HW1, why does the server maintain the game board instead of having individual clients store it? What could go wrong if each client independently maintained own copy?

Client only needs to worry about parsing user input and sending move requests, does not have to validate the moves or maintain the game state. The server keeps the state consistent and enforces the rules. If a client connection fails, no game data is lost. Client could disagree on the game state if updates are missed, so would need a synchronization mechanism

In HW1, suppose you modify the server to create a new thread for every client connection. Two clients simultaneously attempt to place the first X in the same empty cell. What race condition could occur? How would you prevent it?

There would be a race condition that both moves would be considered valid, since the cell is empty at the time each thread is making the request, and both threads would modify the state and return success. We could prevent this with a mutual exclusion lock to prevent both threads from modifying the game board at the same time, which would lead to lost updates.

Why do we use TCP sockets in HW 1 instead of UDP? Explain 2 properties of TCP that make it suitable for this application. Does TCP guarantee that one send() corresponds to one recv()?

TCP has guaranteed delivery and data ordering. This means we are guaranteed to receieve our most requests and game board responses and the data will be in order (important for game moves because order affects the outcome and validity of a move)

One send may be received over multiple recv calls, and multiple sends may be combined into a single recv, so explicit message framing is needed.

TCP vs UDP and what is each good for

TCP: connection-oriented (3-way handshake verifies connection before sending real data), guaranteed delivery (lost packets are resent), data ordering (packets arrive in correct order), slower, congestion control

• good for: tasks where all bytes should be read (downloading app, viewing bank statement)

UDP: connectionless (no handshake), no guaranteed delivery (no retries), no data ordering (packets may arrive scrambled), faster with minimal delay, fixed header size, no congestion control

• good for: real-time streams where frozen screen is worse than tiny glitch (zoom call, movie streaming)

In HW 1, we used 4-byte unsigned integer to prefix every serialized message. What would the prefix be for a serialized object that is 300 bytes long? Why must the receiver read the prefix in loops rather than using one recv() for each?

0x000012C receiver must keep reading until exactly four prefix bytes and 300 payload bytes are read. no guarantee that whole message is in one recv

In HW1, when the server sends the board to the client, should it send the board as data or a string?

It sould send it as data (serialized python object) because the transfer is more efficient, and we can more easily check the length of the message to ensure no data is lost.

computer network

links nodes and enables electronic communication between nodes. may be built on physical or wireless links.

What are the network protocol layers that make communication over network possible?

1) physical 2) data link 3) network ** 4) transport ** -- logical -- 5) session 6) presentation 7) application

network layer

protocols for packet routing through a computer network based on IP address in header, and checksums for data header integrity

Internet Protocol (IP) & properties

route messages from source to destination based on IP address in header

• encapsulates data in datagrams (fragmentation and reassembly)

• best-effort delivery (unrealible)

• connectionless protocol

• data corruption, packet loss, deplication possible

• routing is dynamic (each packet could take a different path and may be out of order)

• checksum to indicate that packet header is not corrupted

IPv4 vs. IPv6

4 uses 32 bit address == 4B addresses

6 uses 128 bit address == 420 undecillion addresses

transport layer

abstraction layer for end-end communication between application processes running on different host machines protocols for directly supporting applications such as those that establish reliable communication (TCP) or support real-time streaming of data (UDP)

Transmission Control Protocol (TCP)

reliable stream delivery service, guarantees that all bytes received will be identical and in the same order as those sent (reliable, ordered, error-checked) also has flow control, congestion control, timeout based retransmission

User Datagram Protocol (UDP)

connectionless communication model with minimal protocol mechanisms. provides integrity verification (checksums) of header and payload. provides no other guarantees to upper layer protocol.

• stateless

• transaction-oriented (simple query and response e.g. DNS)

• lacks of transmission delays (good for real-time)

• multicast makes it useful for broadcasting information

What is a TCP segment?

TCP accepts data from the data stream, divides it into chunks, creates header for each TCP segment

What protocol is connection-oriented? Why?

TCP - uses a 3-way handshake to establish connection. messages include sequence numbers and are acknowledged

how are messages assembled in the OSI model?

each layer adds own control information (headers) to the data as it moves from layer 7 down to layer 1 abstract away what you don't need to know at each layer (efficiency)

RPC

application-layer protocol that provides familiar procedural call model so that calls look like they are happening locally.

biggest challenge of RPC

marshaling parameters - so two sides can understand contents of procedure call. server and client might have diff data representations (e.g. byte order)

RPC process

1) agree on message format

2) agree on representation (integers, characters, endianness)

3) agree on communication (protocols - TCP/IP most common)

Sockets

communication endpoint that an application can write/read data that will be sent/received from the network.

ZeroMQ

abstracts low level socket implementations

• has same interface irrespective of underlying OS

• TCP-based

• allows many-to-one communication (server can listen to multiple ports)

• all communication is asynchronous (sender and receiver do not assume other is online)

• request-reply pattern

Request-reply pattern

traditional client-server communication. client uses request socket of type REQ to send message to server, server uses reply socket of type REP

publish-subscribe zeroMQ

client subscribes to messages that are published by server - server defines topic that client can filter on

• zeroMQ supports multicast from server to many clients

  • server uses PUB socket, client uses SUB socket

Message Passing Interface (MPI)

message passing on parallel architectures

• highly scalable for building large-scale distributed apps

• optimized for cluster interconnects

• little fault tolerance

• requires specialized MPI executors

message-oriented middleware (MOM)

support asynchronous communication using intermediate storage

• sender and receiver dont need to be online at same time

• message delivery has larger latency (mins-hrs)

• persistence communication through middleware-level queues (buffers)

• messages forwarded over series of connected servers

• message guaranteed to insert into dest queue but no guarantee for when it will be read - message stays in queue until removed

queue manager

responsible for routing messages that contain the destination address (logical, location independent names) to the right message queue

challenges of MOM

• all participants must speak the same language, but usually need MOM for applications that are different

• hard to agree on a common language

• need a message broker to translate at