CoE 167 Distributed System Midterms PDF

CoE 167 Distributed System 1. Centralized Solutions cannot scale. Chapter. 1: Introduction Logical and Physical Designs 1.1 Distributed System Logically centralized Solutions – highly scalable...

CoE 167 Distributed System 1. Centralized Solutions cannot scale. Chapter. 1: Introduction Logical and Physical Designs 1.1 Distributed System Logically centralized Solutions – highly scalable distributed (ex. Domain Name System (DNS)) Distributed vs Decentralized - 13 roots servers : not a single point of failure 2. Associated with single point of failure. - Easier to manage - Extremely scalable and robust o Cloud based solutions Integrative View: a need to connect existing :: partial failures- some process and resource is (networked) computer systems to each other not operating Expansive View: an existing system required an Perspectives: extension through additional computers. Expansion:: Improve dependability ( one Architecture – common organization, computer fail, one can take over) common styles Process – forms of process (threads, Decentralized system:: processes and resources virtualization, client, servers), server are necessarily spread across multiple backbone computers Communication – facilities that DS - more on integrative view provide to exchange data between - blockchain (distributed ledger), processes, mimicking procedure federated learning calls, high-level message passing Distributed System:: processes and resources Coordination – fundamental are sufficiently spread across multiple coordination tasks , compensate for computers the lack of global clock, mutual - more on expansive view exclusive access to shared resources - Google Mail Naming – resolving a name to the access point of the named entity Content Delivery Networks (CDNs) – Akamai Consistency and replication – : content of an actual website is copied keeping up with updates and spread across various servers of the CDN Fault tolerance – masking failures : streaming services – choosing servers and recovery : content not copied to all servers, only to where it makes sense, sufficiently. Security - ensure authorized access to resources , trust and authentication Network-Attached Storage device (NAS) 1.2 Design Goals : storage system – file server : users will work on files locally, while also 4 important goals/ requirements :: resources directly accessible by and for other users. easily accessible, hide the fact that Misconceptions: Are centralized solutions bad?  resources are distributed across a network, Relocation: important in cloud computing, should be open, should be scalable services are provided by huge collection of remote servers 1. Resource Sharing Groupware – software for Migration :: mobility of objects , calling, online collaboration tracking BitTorrent Replication:: lockstep mode – no one can take Cloud-based shared storage and files over when another fails Peer-to-peer assisted multimedia Concurrency:: does not notice that the other is streaming making use of the same resource 1Shared mail services (think of outsourced mail systems) Failure:: distinguishing between dead and a slow Shared Web hosting (think of content process, Busy Web servers. distribution networks) Degree of Distribution Transparency 2. Distribution Transparency Aiming at full distribution transparency may be too much Transparent - hide the fact that its processes and resources are physically - There are communication latencies distributed across multiple computers, that cannot be hidden possibly separated by large distances. - Completely hiding failures of networks and nodes is (theoretically and - invisible to end users and applications practically) impossible - middleware o You cannot distinguish a slow computer from a failing one o You can never be sure that a server actually performed an operation before a crash - Full transparency will cost performance, exposing distribution of the system o Keeping replicas exactly up-to- date with the master takes time o Immediately flushing write operations to disk for fault tolerance Exposing distribution may be good Making use of location-based Access: agreement on how data is represented services (finding your nearby in different machines and OS friends) Location: assigning logical names to resources, When dealing with users in Uniform Resource Locator (URL) different time zones  When it makes it easier for a user documents are stored and for how long (i.e., a to understand what’s going on policy) (when e.g., a server does not Self-configurable systems - observes its own respond for a long time, report it as usage and dynamically changes parameter failing). settings Distribution transparency is a nice goal, but 4. Dependability - the degree that a achieving it is a different story, and it should computer system can be relied upon to often not even be aimed at. operate as expected 3. Openness Open distributed system A system that offers components that can easily be used by, or integrated into other systems. An open distributed system itself will often consist of components that originate from Availability : instant :: Reliability :interval elsewhere. High availability:: High Maintainability What are we talking about? Mean Time To Failure (MTTF): The average time Be able to interact with services from other open until a component fails. systems, irrespective of the underlying Mean Time To Repair (MTTR): The average time environment: needed to repair a component. Systems should conform to well- Mean Time Between Failures (MTBF): Simply defined interfaces (Interface MTTF + MTTR. Definition Language) Failure – it cannot meet its promises Systems should easily Error – part of the system that may lead to interoperate – coexist and work failure together Fault – cause of the error Systems should support portability of applications – Transient – occur once and disappear extent on which a DS can be Intermttent – occurs, then vanishes, then executed, w/o modification reappears, so on. Systems should be easily Permanent – continues to exist until faulty extensible – add parts / replace a component replaced file system , plug-ins Ex. Failure – Crashed program Implementation: Error – bug Fault – programmer Policies and Mechanism : Web Caching A browser should ideally provide facilities for 5. Security only storing documents (i.e., a mechanism) and Confidentiality – information disclosed at the same time allow users to decide which only to authorized parties  Integrity – alterations to assets can only 1. File-sharing systems (based, e.g., on be made in an authorized way BitTorrent) 2. Peer-to-peer telephony (early versions of Authorization – checking if an identified Skype) indentity has proper access rights 3. Peer-assisted audio streaming (Spotify) Authentication – verifying correctness of *End users collaborate not administrative claimed identity entities Trust - one entity can be assured that another Techniques for scaling will perform particular actions according to a specific expectation Hide communication latencies (Geographical) Example. Cryptography 1. Make use of asynchronous - encrypting and decrypting using security communication – do something else keys while waiting 2. Have separate handler for incoming Symmetric Cryptosystem – Encryption and response decryption uses single key K Problem: not every application fits this Asymmetric (public-key system) – different key model, eg interactive app (Filling up forms) public key (PK) and secret key (SK) Partitioning and distribution 6. Scalability Move computations to clients Size : add more users and resource (Java applets and scripts) Causes for Centralized : Computational Decentralized naming services capacity, storage capacity, network (DNS) Decentralized information systems Geographical: users and resources lie far apart (WWW) Problems: Synchronous communication, Replication(Owner based) and Caching (Client client blocks until server replies – latency, wide- based): Make copies of data available at area network (WAN) are less reliable than LAN- different machines due to limited bandwidth, limited multipoint Replicated file servers and databases communication Mirrored Websites Web caches (in browsers and proxies) Administrative: easily managed even with many File caching (at server and client) independent admin orgs *Lead to consistency problems Problems: policy conflicts 1.3 Simple Classification Sharing expensive equipment in Computational 1. High Performance Distributed grids– allows a computer at org A to directly Computing access resources at org B Counterexample:  Multiprocessor machines – all have access to Durable – commit means changes are same physical memory (for improving the permanent performance of programs, easy to program) Nested transactions Multicomputer system – computers have their own memory Distributed shared-memory multicomputer (DSM system) – allows a processor to address a memory of another computer as if it were local Cluster computing – collection of similar compute nodes, for parallel programming, homegeneity Example: Planning a trip which three different Grid Computing – decentralized systems flights need to be reserved that are often constructed as a federation of computer systems, nodes are specifically configured for certain tasks, collaboration in form of virtual organization 2. Distributive information systems Integration: A server running an application making it available to remote programs (client). Transaction Processing Monitor (TP Monitor) – Client request -> response is sent. At lowest responsible for coordinating the execution of a level, wrap requests into a single larger request transaction, allow an app to access multiple (distribution transaction) , all or none of the server. Distributed Commmit – protocol request are executed. Enterprise Application Integration (EAI) – applications communicate directly with each other Transactions – operations on a database Middleware as communication facilitator in EAI ACID properties Types: Atomic – happens indivisibly Remote Procedure Calls (RPC) - Requests are Consistent – does not violate system invariant sent through local procedure call, packaged as Isolated – do not interfere with each other message, processed, responded through and devices is highly unobtrusive, hide message, and result returned as return from call. interfaces (implicit action) 3. (Context awareness) The system is aware Message-oriented middleware (MOM) - of a user’s context to optimize interaction, Messages are sent to logical contact point shared data space – processes decoupled in (published), and forwarded to subscribed time and space are attracive applications. 4. (Autonomy) Devices operate Integrating Application: autonomously without human intervention, and are thus highly self-managed 1. File transfer : simple but not flexible 5. (Intelligence) The system as a whole can a. Figure out file format and layout handle a wide range of dynamic actions and i. XML (extended markup interactions language) – files are self describing 2. Mobile computing systems: pervasive, b. Figure out file management but emphasis is on the fact that devices are c. Update propagation, and update inherently mobile. notifications Features: 2. Shared database: much flexible a. Need to design common data A myriad of different mobile devices scheme (smartphones, tablets, GPS devices, remote b. Performance bottleneck (too many controls, active badges). read and update) Mobile implies (wireless) that a device’s 3. Remote Procedure calls - Effective when location is expected to change over time ⇒ execution of a series of actions is needed. change of local services, reachability, etc. 4. Messaging - RPCs require caller and Keyword: discovery. callee to be up and running at the same Maintaining stable communication can time. Messaging allows decoupling in introduce serious problems. time and space. For a long time, research has focused on 3. Pervasive Systems directly sharing resources between mobile Intended to blend with devices. MANETS(mobile ad hoc networks). It environment naturally never became popular and is by now considered to be a fruitless path for research. Subtypes: 1. Ubiquitous computing systems: pervasive and continuously present, i.e., there is a continuous interaction between system and user. Core requirements: 1. (Distribution) Devices are networked, distributed, and accessible transparently 2. (Interaction) Interaction between users  Essence: a mobile device connected to a server (and nothing else) [making use of cloud-based services] First: sensors do not cooperate but simply send data to a centralized database at the operator site Mobile Edge Computing (MEC) – latency Second: forward queries to relevant sensors and and computational issues (ex. Augmented let each other compute an answer, operator reality, interactive gaming, etc), immediate aggregates the response feedback Cloud-edge continuum 4. Sensor (and actuator) networks: pervasive, with emphasis on the actual (collaborative) sensing and actuation of the environment. The nodes to which sensors are attached are: Pitfalls Many (10s-1000s) | Simple (small memory /compute/ False assumptions that many make when communication capacity) developing a distributed application for the first Often battery-powered (or even battery- time: less) The network is reliable Ex. Activation of sprinklers when fire is The network is secure detected The network is homogeneous The topology does not change Latency is zero Bandwidth is infinite Transport cost is zero There is one administrator Chapter 2: Architecture Service- functionality; what the service does Interface- specifies how higher entities access the service – hides the implementation Protocol – describe the rules to exchange information Ex. TCP Two-party communication Service: reliable communication over a network Interface: Socket 2.1 Architectural Styles Protocol: Tranmission Control Protocol - formulated in terms of components :: a Application Layering: PAD Model modular unit with well-defined interfaces that is replaceable within its environment: Application – interface layer (Presentation layer) contains units for interfacing to users or external The way they are connected applications [you sitting in front of computer, and about The data exchanged b/w them to use a browser] How these elements are jointly configured Processing layer - contains the functions of an application, i.e., without specific data [ When you Connector – a mechanism that mediates send a search request to somewhere, it does a core communication, coordination, and cooperation functionality of a search. !!it will rely on other services that among components. Ex. facilities for (R)PC, will hold the data.] message passing, or streaming data. Allow for Data Layer - contains the data that a client wants the flow of control and data to manipulate through the application 2.1.1 Layered Architectures components (a) Pure layered organization. (b) Mixed layered organization. (c) Layered organization with upcalls a) Common in network communication b) Using libraries for applications c) E.g. OS signals the occurrence of an event a. It has to do with handle – Its possible to subscribe to events, and when they become available, it gets an automatic notification. Example:: Search engine b. This case, n-1 is interested in event in n-2, so n-2 notifies n-1 that the (UI) User types in a search keyword → (Processing) event (handle) happened, using an Query generator needs db → (Data) db query is upcall. taken care of → (Data) the info (page titles and info) is sent back to processing level → (processing) All Layered Communication Protocol results are ranked → (processing) generates HTML pages→ (UI) its passed back to UI level.  Separation between interfaces and the objects Layered architecture implementing these interfaces allows us to place an interface at one machine, while the - Very popular object itself resides on another machine. - Drawback: strong dependency b/w different layers if not well designed Good ex. – Communication protocol stacks Bad ex. - applications designed and developed as compositions of existing components without much concern for the stability of interfaces nor overlap of functionality between different components Client-side stub (proxy) 2.1.2 Service Oriented an architectural style reflecting a more 1. Client invokes a method loose organization into a collection of 2. Server gives a copy of interface to client → separate, independent entities. Each called proxy. Proxy is loaded into clients entity encapsulates a service address space. the service is executed as a separate o Proxy marshal the method invocation process (or thread) client made can be called service or object o and unmarshal reply messages to return the result of the method invocation to the client. Object-based style 3. The marshalled invocation is passed across network Objects corresponds to components, Server-side stub (skeleton) connected and communicate through a procedure call. 1. Incoming invocation requests are first sent If two objects reside on the same system to a server stub (skeleton) → method call, over a network → remote o Which unmarshals the invocation client procedure call. sent, and actually make method invocation that client wants at the server through an interface. o Also creates a reply and marshals them and forwards reply msgs to the client- side proxy - Proxy and the skeleton are referred to as stubs!!! BOARD Object encapsulate the data (state), as they exhibit Resource-based architecture the interface, but never shows how its implemented Encapsulation - Objects are said to encapsulate View a distributed system as a set of data and offer methods on that data without resources where machines, individually revealing the internal implementation. managed by components. Resources may be added, deleted, modified, etc by Remote Object (remote) apps. Representational State Transfer (REST) : RESTful 2.1.3 Publish-subscribe architecture As processes join and leave, its important that dependencies btwn processes are as 1. Resources are identified through a single loose as possible → hence, architecture that naming scheme. usually accessed has strong separation between processing through URI (uniform resource identifier) and coordination 2. All services offer the same interface. e.g. o So more of autonomously operating put, get, delete, post processess 3. Messages sent to or from a service are Coordination fully self-described. E.g. when sending encompasses the communication and HTML, say that it is HTML. Send its media type. cooperation between processes 4. After executing an operation at a service, glue that binds the activities performed by that component forgets everything about processes into a whole the caller (stateless execution). a. Once the sever gets the request, that server Coordination Models takes the rq, process it, sends back the resource and forgets it → is memoryless execution Temporal coupling – process need to be b. Is prominent in web services and REST available at the same time Operations: CRUD operation Referential coupling – process need to Create (PUT), Read (GET), Update (POST), Delete know the name or identifier of other (DELETE) process Design: Example: Amazon’s Simple Storage Service 1. Mailbox – Temp decoup, Ref Coup- 2 processors, to work and exchange info – they use Objects (=files) are placed shared mailbox , and they communicate through into buckets (=directories) this shared mailbox a. Write and fetch to the mailbox By placing a file in a bucket, file is b. No real communication btwn the two automatically uploaded to the Amazon cloud Event-based – Temp Coup, Ref Decoup - ObjectName contained in BucketName – coordination btwn processors will happen once the event access through: occurs http://BucketName.s3.amazonaws.com/Obje Processer 1 publish a notification describing the ctName occurrence of event 1, and if ur interested, URI - Operations are carried out by sending you subscribe, and will be notified!! And will have access HTTP requests – to it Publish and subscribe o PUT - request through HTTP o GET – to see if the object is contained in Event-based and Shared-data spaec the BucketName o S3 – access a file in s3 web service o Specific object in that bucket called ObjectName  Facilities for inter-application communication. Security services. Accounting services. Masking of and recovery from failures. 2.2.1 Middleware organization 1. Event based architectural style – publish Problem The interfaces offered by a legacy subscribe is key. component are most likely not suitable for all o Event bus – mechanism which the applications. publishers and subscribers are matched → what coordinates these events Solution A wrapper or adapter offers an 2. Shared data space architectural style – interface acceptable to a client application. Its there’s a db which is persistent and liable functions are transformed into those available at o The components will communicate the component. entirely through tuples which is saved in a saved db, and other one does a quick search to see if the tuple exists any tuple that matches is returned. o Tuples: a structured data records with number of fields - Can be combined w event based – process subscribes to certain tuples publish &subscribe o Publish and subscibe : event match Interceptors Events are (attiribute, value) pair 1. Topic based subscription : attribute = value - a software construct that will break 2. Content based subscription: attribute ∈ range the usual flow of control and allow other (application specific) code to be executed The principle of exchanging data items between - primary means for adapting publishers and subscribers. middleware to the specific needs of an application 2.2 Middleware and distributed systems Commonly used components and functions that need not be implemented by applications separately: 2.3 Layered-system Architecture 2.3.1 Simple client-server architecture o B – entire UI on the client side, which One of the core organization of the system communicates with rest of the app architecture through a protocol on the server. Server – process implementing a specific Client does no processing service – e.g. file or db service o C –processing is partly done in the Client – a process that requests a service client side. Front end checks for the from server correctness of the form, and more The way it works: Clients and servers could complex in server. be in a diff machine – both follows request- o D – client is not only in charge of UI, reply model but the processing! But for data, o Client sends the request, server accepts, server does the service, server replies → gotta ask the server would take a bit of time o By means of a simple connectionless protocol- efficient. But making the protocol resistant to transmission failure is not trivial o E – client’s local disk has partly some data → doing a lot → FAT client!! Not good lmao ▪ E.g. web browsing – build huge 2.3.2 Multitiered Architectures cache on local disk 1. Single tiered: dumb terminal/mainframe If we have more functionalities on the configuration client side, it has to be more end-user 2. Two-tiered: client/single server configuration’ resilient, and has to cope with a lot of different platform, needing for a multiple version- hence not optimal. UI is split btwn the client and server o E.g. if ur in front of the machine and 3. Three-tiered: each layer on separate filling in the form - it can be machine implemented on both server or client side prev lec 5 diff ways to implement PAD model o A - only terminal dependent UI is in client side, but also exists in server side.  Client -> Application -> Database. HTML files could be referred to each other Processing layer are executed by a by a hyperlink separate server A Web server essentially needed only a hyperlink to fetch a file A browser took care of properly rendering the content of a file Ex. Less simple Web servers o UI – sends request to the application and present the result to the user o Application server – doesn’t have the data although it understands the A website was built around a database business logic. Get the data from db with content server and process it, and send it up A Webpage could still be referred to by a o Database server – take the rq, hyperlink process and send it to application A Web server essentially needed only a layer hyperlink to fetch a file E.g. transaction processing. A separate A separate program (Common Gateway process, a transaction processing Interface) composed a page monitor, coordinates all transactions across different data servers A browser took care of properly rendering the content of a file Ex. The Network File System (NFS) Each NFS server provides a standardized 2.4 Symmetrically distributed system view of its local file system: each server architectures supports the same model, regardless the Alternative organizations implementation of the file system. 1. Vertical distibution - comes from dividing NFS Architecture distributed applications into three logical Ex. Simple Web Servers layers, and running the components from each layer on a different server (machine) 2. Horizontal distribution – a client or server may be physically split up into logically equivalent parts, but each part is operating on its own share of the complete data set. 3. Peer-to-peer architecture - Processes are all equal: the functions that need to be A website consisted as a collection of carried out are represented by every HTML files process ⇒ each process will act as a  client and a server at the same time (i.e., The ring is extended with various shortcut acting as a servant links to other nodes. o They communicate, and could join and leave, but the interaction is always there. o There is no master node – processes are all equal → each process will act as a client AND server o Sometime process will need functionality hence would have to send the request 2.4.1 Structured peer-to-peer Nodes are organised in an overlay with specific topology. E.g. binary tree, grid → for 2.4.2 Unstructured peer-to-peer data lookup Each node maintains an ad hoc list of o Based on using semantic-free index: neighbors. The resulting overlay Each data item is associated with a key, resembles a random graph: an edge ⟨u,v⟩ which is used as an index. Its common to exists only with a certain probability use a hash function. Key (data item) = hash P[⟨u,v⟩]. (data item’s value) o each node responsible for storing (key, Searching value) pairs - thus implementing a distributed hash table 1. Flooding: issuing node u passes request Any node can be asked to lookup a given key , for d to all neighbors. Request is ignored which then comes down to efficiently routing when receiving node had seen it before. that lookup request to the node responsible for Otherwise, v searches locally for d storing the data associated with the given key (recursively). a. May be limited by a Time-To-Live: Example: Chord a maximum number of hops. 2. Random walk: : issuing node u passes Nodes are logically organized in a ring. request for d to randomly chosen Each node has an m-bit identifier. neighbor, v. If v does not have d, it Each data item is hashed to an m-bit key. forwards request to one of its randomly Data item with key k is stored at node with chosen neighbors, and so on. smallest identifier id ≥ k, called the Random walks are more communication successor of key k. efficient (require less resource), but might take longer 2.4.3 Hierarchically organized peer-to- peer networks Super-peer networks  When searching in unstructured P2P Chapter 3: Process systems, having index servers improves 3.1 Threads performance Deciding where to store data can often be Processor: Provides a set of instructions along done more efficiently through brokers with the capability of automatically executing a series of those instructions. Thread: A minimal software processor in whose context a series of instructions can be executed. Saving a thread context implies stopping the current execution and saving all the data needed to continue the execution at a later stage Example. BitTorrent Process: (a program in execution) a software processor in whose context one or more threads may be executed. Executing a thread, means executing a series of instructions in the context of that thread. Virtual processors: : programs should be able to Lookup file at a global directory ⇒ returns share the CPU without one program halting a torrent file progress of the others Torrent file contains reference to tracker: Threads allow multiple programs to share a CPU a server keeping an accurate account of active nodes that have (chunks of) F. 3.1.1 Introduction to threads P can join swarm, get a chunk for free, and Thread – basic unit of CPU utilization, consists then trade a copy of that chunk for another of a program counter, a stack, and a set of one with a peer Q also in the swarm. register, (and a thread ID) “tit-for-tat” mechanism - reward good behavior Traditional processes have single thread of control, one program counter, one sequence of instructions Multi-threaded applications Multiple threads within a single process Each thread has own program counter, stack and set of registers Threads share common code, data, and certain structures such as open files. (making files available, fast upload/download speed) Process abstraction: virtual computer Why use Threads makes the program feel like it has the Avoid needless blocking: a single- entire machine to itself – like a fresh threaded process will block when doing computer has been created, with fresh I/O; in a multithreaded process, the memory, just to run that program operating system can switch the CPU to another thread in that process. Thread abstraction: virtual processor Exploit parallelism: the threads in a Simulates making a fresh processor multithreaded process can be scheduled inside the virtual computer represented to run in parallel on a multiprocessor or by the process multicore processor. This new virtual processor runs the same Avoid process switching: structure large program and shares the same memory applications not as a collection of as other threads in the process processes, but through multiple threads. Context Switching Processor context: The minimal collection of values stored in the registers of a processor used for the execution of a series of instructions (e.g., stack pointer, addressing registers, program counter). Thread context: The minimal collection of Trade-offs values stored in registers and memory, Threads use the same address space: used for the execution of a series of more prone to errors instructions (i.e., processor context, state). No support from OS/HW to protect Process context: The minimal collection of threads using each other’s memory values stored in registers and memory, used Thread context switching may be faster for the execution of a thread (i.e., thread than process context switching context, but now also at least MMU register The cost of context switching values). Direct costs: time for actual switch and Observations executing code of the handler 1. Threads share the same address space. Indirect costs: other costs, notably caused by Thread context switching can be done messing up the cache entirely independent of the operating system. 2. Process switching is generally (somewhat) more expensive as it involves getting the OS in the loop, i.e., trapping to the kernel. 3. Creating and destroying threads is much cheaper than doing so for processes.  Concurrency When there are more threads than processors, concurrency is simulated by time slicing processor switches between threads *If join is removed in line 21 then print function in line 22 will immediately execute On most systems, time slicing happens unpredictably and nondeterministically, meaning that a thread may be paused or resumed at any time Threads and OS Main issue: Should an OS kernel provide threads, or should they be implemented as user-level packages? User-space solution All operations can be completely handled within a single process ⇒ * Threads share same variable (shared_x) implementations can be extremely efficient. All services provided by the kernel are done on behalf of the process in which a thread resides ⇒ if the kernel decides to block a thread, the entire process will be blocked. Threads are used when there are many external events: threads block on a per- event basis ⇒ if the kernel can’t distinguish threads, how can it support signaling events to them? Kernel solution The whole idea is to have the kernel contain the implementation of a thread package. This means that all operations return as system calls:  Multithreaded web client Operations that block a thread are no Hiding network latencies: longer a problem: the kernel schedules Web browser scans an incoming HTML another available thread within the page, and finds that more files need to be same process. fetched. handling external events is simple: the Each file is fetched by a separate thread, kernel (which catches all events) each doing a (blocking) HTTP request. schedules the thread associated with As files come in, the browser displays the event. them. The problem is (or used to be) the loss of efficiency because each thread Multiple request-response calls to other operation requires a trap to the kernel. machines (RPC) Combining user-level and kernel level threads A client does several calls at the same time, each one by a different thread. It then waits until all results have been returned. Note: if calls are to different servers, we may have a linear speed-up. Thread-level parallelism: TLP ∑𝑁 𝑖=1 𝑖 ∙ 𝑐𝑖 Kernel threads that can execute user-level 𝑇𝐿𝑃 = 1 − 𝑐0 threads 𝑐𝑖 : fraction of time that exactly 𝑖 threads are User thread does system call ⇒ the being executed simultaneously kernel thread that is executing that user thread, blocks. The user thread remains 𝑁: maximum number of threads that can bound to the kernel thread. execute at the same time The kernel can schedule another kernel Practical measurements: thread having a runnable user thread bound to it. Note: this user thread can A typical Web browser has a TLP value switch to any other runnable user thread between 1.5 and 2.5 ⇒ threads are primarily currently in user space. used for logically organizing browsers. A user thread calls a blocking user-level 3.1.2 Threads in distributed systems operation ⇒ do context switch to a runnable user thread, (then bound to the Using threads at server side same kernel thread). Improve performance When there are no user threads to schedule, a kernel thread may remain Starting a thread is cheaper than starting idle, and may even be removed a new process. (destroyed) by the kernel. Using threads at the client side  Having a single-threaded server prohibits 1. Instruction set architecture:interface simple scale-up to a multiprocessor system. As with clients: hide network latency by reacting to next request while previous one is being replied. Better structure b/w hardware and software, the set of Most servers have high I/O demands. machine instructions, with two subsets. Using simple, well-understood blocking ▪ Privileged instructions: allowed to be calls simplifies the structure. executed only by the operating Multithreaded programs tend to be system. smaller and easier to understand due to ▪ General instructions: can be executed simplified flow of control. by any program. 2. System calls as offered by an operating Multithreading organization system. Dispatcher/worker model 3. Library calls, known as an application programming interface (API) Ways of Virtualization (a) Separate set of instructions, an interpreter/emulator, running atop an OS. 3.2 Virtualization (b) Low-level instructions, along with bare-bones Hardware changes faster than software minimal operating system Ease of portability and code migration (c) Low-level instructions, but delegating most Isolation of failing or attacked work to a full-fledged OS. components VMs vs Containers 3.2.1 Principle of virtualization - extending or replacing an existing interface to mimic the behavior of another system Mimicking Interfaces Four types of interfaces at three different levels  3.2.2 Containers sharing a physical machine with other Sits on top of host OS, thus only OS is customers ⇒ almost complete virtualized isolation between customers Each container shares the host OS kernel, (although performance isolation may the binaries and libraries not be reached). Shared components are read-only Example: AWS Amazon Web Service EC2: Amazon Elastic Compute Cloud allows one to create an environment consisting of several networked virtual servers, thus jointly forming the basis of a distributed system AMI: Amazon Machine Images large number of installable software packages consisting of an operatingsystem kernel along with several Namespaces: a collection of processes services in a container is given their own view of example: LAMP image: Linux (kernel), identifiers Apache (Web Server), MySQL (database Union file system: combine several file system), PHP (libraries) systems into a layered fashion with only An AMI, when launched, is called an EC2 the highest layer allowing for write instance: the actual virtual machine that operations (and the one being part of a can be used to host the customer’s container). Think of directories applications Control groups: resource restrictions can local storage that comes with an instance be imposed upon a collection of is transient: when the instance stops, all processes the data stored locally is lost 3.2.3 Application of VMs to DS Persistent Storage (prevent data loss) Three types of cloud services Elastic Block Store (EBS) – mounted as 1. Infrastructure-as-a-Service covering one would mount a hard disk, typically the basic infrastructure accessible to one EC2 instance at a time 2. Platform-as-a-Service covering Simple Storage Service (S3) –accessible system-level services to multiple instances or other cloud 3. Software-as-a-Service containing services actual applications IAAS - Instead of renting out a physical machine, a cloud provider will rent out a VM (or VMM) that may be  3.3 Clients 3.3.1 Networked user interfaces Client Server Interaction Thin-client approach – everythin is processed and stored at the server. Ex. The X window system (not Twitter) 3.3.3 Client-side software for distribution - Used to control bit-mapped terminals transparency (monitor, keyboard, mouse) Access transparency: client-side stubs for RPCs Location/migration transparency: let client-side software keep track of actual location Replication transparency: multiple invocations handled by client stub: X client and server The application acts as a client to the X-kernel, the latter running as a server on the client’s machine. Failure transparency: can often be placed 3.3.2 Virtual desktop environment only at client (we’re trying to mask server Logical development and communication failures). With an increasing number of cloud-based applications, the question is how to use those 3.4 Servers applications from a user’s premise? Issue: develop the ultimate networked 3.4.1 General design issues user interface Server - A process implementing a specific Answer: use a Web browser to establish service on behalf of a collection of clients. It a seamless experience waits for an incoming request from a client Ex. The Google Chromebook and subsequently ensures that the request is taken care of, after which it waits for the next incoming request.  Issue 1: Iterative vs Concurrent Urgent message comes in ⇒ associated request is put on hold Iterative server: Server handles the Note: we require OS supports priority- request before attending a next request. based scheduling Concurrent server: Uses a dispatcher, Solution 2: Use facilities of the transport layer which picks up an incoming request that is Example: TCP allows for urgent messages then passed on to a separate thread/process. in same connection Multitthreaded servers. Urgent messages can be caught using OS * Concurrent servers are the norm: they can signaling techniques easily handle multiple requests, notably in Issue 4: Stateless vs Stateful Server the presence of blocking operations (to disks or other servers). Stateless server: Never keep accurate information about the status of a client after Issue 2: Contacting a server having handled a request: Services are tied to a specific port (end Don’t record whether a file has been point) opened (simply close it again after access) Don’t promise to invalidate a client’s cache Don’t keep track of your clients Consequence: Server and corresponding TCP port Clients and servers are completely Assigning an end point (Daemon and independent Superserver) State inconsistencies due to client or server crashes are reduced Possible loss of performance because, e.g., a server cannot anticipate client behavior (think of prefetching file blocks) Issue 3: Interrupting a server Stateful server: Keeps track of the status of its clients: Out of band communication Record that a file has been opened, so Issue: Is it possible to interrupt a server once it that prefetching can be done has accepted (or is in the process of accepting) a Knows which data a client has cached, service request? (file transfer then cancelling and allows clients to keep local copies of mid way) shared data Solution 1: Use a separate port for urgent data Server has a separate thread/process for * The performance of stateful servers can be urgent messages extremely high, provided clients are allowed to keep local copies. As it turns out, reliability is First tier: request dispatching - passing requests often not a major problem. 3.4.2 Object servers Objects: data (state) & code (method) object server itself does not provide to an appropriate server *Having the first tier handle all communication from/to the cluster may lead to a bottleneck. First tier: consists of switch through which clients are routed service, but invokes local objects that provide the service upon request from Second tier: application layer switch , inspect clients content of requests Activation policy: which actions to take Third tier: data-processing server when an invocation request comes in: o Where are code and data of the Solution: TCP Handoff object? Idea: Switch (decides which server) handoff o Which threading model to use? connection to server and all responses are o Keep modified state of object, if directly communicated to the client without any? passing through the switch. Object adapter: implements a specific Wide area (WAN) server Cluster activation policy Spreading servers across the Internet may introduce administrative problems. These can be largely circumvented by using data centers from a single cloud provider. Request dispatching : if locality is important Common approach: DNS 1. Client looks up specific service through DNS - client’s IP address is part of request 2. DNS server keeps track of replica servers 3.4.3 Server clusters for the requested service, and returns address of most local server Local area (LAN) Server clusters Client Transparency Three-tiered server cluster To keep client unaware of distribution, let DNS resolver act on behalf of client. Problem is that the resolver may actually be far from local to the actual client. Simplified Akamai CDN * The cache is often sophisticated enough to hold more than just passive data. Much of the application code of the origin server can be moved to the cache as well. Step 1: Client looks up regular domain name, redirected to the akamai.net resolvers. Step 2: Resolvers will look up best edge server to serve the client and return its network address. Step 3: Allows client to contact edge server Step 4: If request is not in edge server’s cache, will fetch in the origin server , caches and returns the requested one to the client. 3.5 Code migration 3.5.1 Reasons for migrating code Load distribution o Ensuring that servers in a data Example: federated machine learning center are sufficiently loaded (e.g., to prevent waste of energy) 3.5.2 Models for code migration o Minimizing communication by ensuring that computations are close to where the data is (think of mobile computing). Flexibility: moving code to a client when needed Dynamically configuring a client to communicate with a server * Avoids pre-installing software and increases dynamic configuration. Privacy and security: if data cannot be CS: C E R (code, exec state, resource segment) moved to another location, move the are all in Server, after execution only E is code to the data updated. REV: (sender-initiated) client migrates code to Only Solution: abstract machine implemented the server where the code is executed, E is on different platforms modified at the server Interpreted languages, effectively having CoD: (receiver-initiated) client obtains code from their own VM server with its execution, modifying E on client Virtual machine monitors side Migrating images: three alternatives MA: sender- initiated – C and E are moved from 1. Pushing memory pages to the new client to server, operating on both client and machine and resending the ones that are server’s resources. later modified during the migration Code segment : contains actual code process. 2. Stopping the current virtual machine; Data segment : contains the state migrate memory, and start the new virtual Execution sate: contains the context of thread machine. executing the objects code. 3. Letting the new virtual machine pull in Weak mobility: Move only code and data new pages as needed: processes start on segment (and reboot execution) - transferred the new virtual machine immediately and program restarts copy memory pages on demand. Problem: A complete migration may actually Relatively simple, especially if code is take tens of seconds. We also need to realize portable that during the migration, a service will be Distinguish code shipping (push) from completely unavailable for multiple seconds. code fetching (pull) Strong mobility: Move component, including execution state - running process can be stopped, moved, and resume exactly where it left off. Migration: move entire object from one machine to the other Cloning: start a clone, and set it in the same execution state. 3.5.3 Migration in heterogeneous systems Chapter 4: Communication Main problem: 4.1 Foundations The target machine may not be suitable to Recap on Low-level Layers execute the migrated code The definition of process /thread 1. Physical Layer: contains the specification /processor context is highly dependent on and implementation of bits, and their local hardware, operating system and transmission between sender and runtime system receiver  2. Data Link : prescribes the transmission of Transport Layer - provides the actual a series of bits into a frame to allow for communication facilities for most distributed error and flow control systems. 3. Network Layer : describes how packets in Standard Internet Protocols a network of computers are to be routed. Transmission Control Protocol (TCP): Basic networking model : Open System connection-oriented, reliable, stream- Interconnection (OSI) Model oriented communication Drawbacks Focus on message-passing only Often unneeded or unwanted functionality Violates access transparency Physical layer Deals with standardizing how two computers are connected and how 0s and 1s (bits) are represented. Data link layer Provides the means to detect and possibly correct transmission errors, as Universal Datagram Protocol (UDP): well as protocols to keep a sender and receiver unreliable (best-effort) datagram in the same pace. communication Network layer Contains the protocols for Middleware Layer - provide common services routing a message through a computer network, and protocols that can be used by many as well as protocols for handling congestion. different applications Transport layer Mainly contains protocols for A rich set of communication protocols directly supporting applications, such as those (Un)marshaling of data, necessary for that establish reliable communication, or integrated systems support real-time streaming of data. o if we are sending the data through network, we have to marshall and Session layer Provides support for sessions unmarshall. U’ll need to build a msg between applications. and tell them what ur gonna do with Presentation layer Prescribes how data is this msg represented in a way that is independent of the 1. Marshalling and unmarshalling itself is an overhead because the hosts on which communicating applications are data is serialized into 1s and 0s, running. and when marshalling, it has to Application layer Essentially, everything else: compose the msgs e-mail protocols, Web access protocols, file- o transfer protocols, and so on. Naming protocols, to allow easy sharing of resources  o In middleware, you are accessing 3. Server connects to the mail delivery resources all over the place. So you system and does the rq - deliver the need some sort of protocol to name message those processes 4. Notify the client that its done - another Security protocols for secure synchronization communication Scaling mechanisms, such as for replication and caching Persistent communication: message for o Middleware should deal with this transmission is stored by the middleware as long transparently!!very imp in terms of at it takes to deliver to the receiver. Ex. e-mail scaling system 1. Since it replicates and caches, it - So not necessary of sending app to continue takes up a lot of data application Transient communication: message is stored by the communication system only as long as Adapted layering scheme the sending and receiving application are executing. - Comm. server discards message when it cannot be delivered at the next server, or at the receiver. - If the app on the other side is unavailable and when you can’t communicate, it discards the message → does not ensure the Session and Presentation is replaced with message sending Middleware - All transport layer typically provides this. Network and Transport grouped into Asynchronous communication: sender communication services offered by the continues immediately after it has submitted its OS message for transmission, message is (temporarily) stored immediately by the MW Types of Communication * Middleware as an additional service in client- server computing 1. First need to synchronize – server and client. 2. Client sends the msg to the mail delivery system 1. Once the msg gets through the server, the message is stored in a upon submission. storage facility. Synchronous communication: sender is Sender need not wait for immediate reply, blocked until request in know to be accepted but can do other things Middleware often ensures fault tolerance Places for synchronization At request submission 4.2 Remote Procedure Call (RPC) At request delivery Application developers are familiar with After request processing simple procedure model Well-engineered procedures operate in isolation (black box) Client/Server – transient synchronous There is no fundamental reason not to execute procedures on separate machine Client and server have to be active at the Thus, communication between caller & time of communication callee can be hidden by using procedure- Client issues request and blocks until it call mechanism. receives reply Server essentially waits only for incoming requests, and subsequently processes More detailed RPC Operation them 1. Client procedure calls client stub. 2. Stub builds message; calls local OS. Drawbacks synchronous communication 3. OS sends message to remote OS. 4. Remote OS gives message to stub. Client cannot do any other work while 5. Stub unpacks parameters; calls server. waiting for reply 6. Server does local call; returns result to stub. Failures have to be handled immediately: 7. Stub builds message; calls OS. the client is waiting 8. OS sends message to client’s OS. The model may simply not be appropriate 9. Client’s OS gives message to stub. (mail, news) 10. Client stub unpacks result; returns to client Messaging – high level of persistent RPC: Parameter passing asynchronous communication There’s more than just wrapping parameters into Message oriented middleware a message Processes send each other messages, Client and server machines may have which are queued different data representations (think of byte ordering)  Wrapping a parameter means transforming a value into a sequence of bytes Different from the traditional RPC, in Client and server have to agree on the asynchronous RPC, client can continue same encoding: without waiting for an answer from the o How are basic data values server!! represented (integers, floats, Calls request → waits for acceptance, characters) and continue doing other thing → when o How are complex data values results are available, server sends a represented (arrays, unions) callback to client But the problem is, reliability is not ** Client and server need to properly interpret guaranteed and we don’t knowif its messages, transforming them into machine- gonna be processed. dependent representations. Client and server does not need to Some assumptions: synchronize constantly, and that you can do smth else Copy in/copy out semantics: while procedure is executed, nothing can be assumed about parameter values. Sending out multiple RPCs : send RPC request All data that is to be operated on is to multiple servers passed by parameters. Excludes passing references to (global) data. ** Full access transparency cannot be realized. A remote reference mechanism enhances access transparency Remote reference offers unified access to remote data Remote references can be passed as parameter in RPCs Note: stubs can sometimes be used as such references 4.3 Message Oriented Communication Asynchronous RPCS : Try to get rid of the strict Transient Messaging: Sockets – comm. end request-reply behavior, but let the client points, app can write data to be sent out and continue without waiting for an answer from the from which incoming data can be read serve. (Deffered ARPC) ↓  Sockets are rather low level and Publish-subscribe senders of messages, called publishers, do not program the messages to be sent directly to specific receivers, but rather are sent to subscribers of specific topics programming mistakes are easily made. However, the way that they are used is often the same (such as in a client-server setting). Alternative: ZeroMQ Higher level of expression : pairing sockets Pipeline (Push/Pull) o One for sending msgs at process P o One at process Q for rcving msgs Distribute messages to multiple workers, All comms. are asynchronous. arranged in a pipeline A Push socket will distribute sent Three patterns messages to its Pull clients evenly Request-reply Results computed by consumer are not sent upstream but downstream to traditional client-server communication, another pull/consumer socket 7 like the ones normally used for remote Data always flows down the pipeline, and procedure calls each stage of the pipeline is connected to o Client application sends a request at least one node to a server and expects the latter to respond with an appropriate response unlike traditional client-server, can connect to many servers o Requests interleaved or distributed to both the servers Message Passing Interface (MPI) When lots of flexibility is needed Transient communication Representative Operations msg that was sent, but not necessary that the sender is executing d) Both sender and receiver are inactive o They are loosely coupled – no need for them to be synchronized as everything goes through a queue!! MOM - Asynchronous persistent communication through support of middleware- level queues. Queues correspond to buffers at communication servers. Message-oriented persistent communication Message-oriented middleware (MOM) or Queue-based messaging offers intermediate-term storage capacity for messages, without requiring either the sender or receiver to be active during Notify: if I’m interested in event that hasn’t arrived message transmission. in the queue, if it arrives notify me! new way of thinking and designing MPI Model – queuing Queue managers : Queues are managed Provides support for Asynchronous by queue managers. An application can put persistent – and middleware does this messages only into a local queue. Getting a by implementing a queue, and neither message is possible by extracting it from a local sender nor receiver has to be active queue only ⇒ queue managers need to route messages. Routing : a) Both sender and receiver is active b) Sender is active but receiver is not active If we take this queue-level addressing – msg can’t be delivered and link it to the network- level o Append msg in the queue and get it addressing, its so similar once they are active c) Send the message and went to do smth We communicate through queue, and else, receiver is active – receiver can read client need to know how the destination queue is addressed. So we need to know  the name of the queue and the address of where it can be accessed. → queue manager Each name is associated with contact address (host, pair) You need lookup that makes the name- to-address be available for queue manger May provide subject-based routing capabilities (i.e., publish-subscribe capabilities) o Lookup says, I’m a sender, I need to General Architecture append this msg, give me the name For each pair of application, we have and the address of that queue. separate subprogram – converts o I want to send a msg to this receiver, messages b/w two apps where is its queue → it then finds the Subprogram is drawn as a plugin – address of the queue, and the msg emphasis that they can be plugged in or will be routed to that queue removed from the broker Is what DNS implements, in the sense that it passes the address when you give Example: AMQP it a name and we have an address Due to lack of standardization -> Advanced lookup database Message-Queueing Protocol (AMQP) was intended to play the same role as, for example, Message broker TCP in networks: a protocol for high-level