CS162 Lecture 25: Distributed Storage, NFS and AFS, Key Value Stores
Watch on YouTubeVideo summary
This lecture explores fundamental concepts in distributed storage, beginning with the mechanics of Remote Procedure Calls (RPC) and the constraints imposed by the CAP theorem on file system design. RPC enables clients to execute functions across a network as if they were local operations, but this requires serializing arguments because machine-specific pointers cannot be shared directly between different systems. To manage data access transparently regardless of whether files reside locally or remotely, Virtual File Systems (VFS) provide an abstraction layer that allows various file system types to be mounted onto the same directory hierarchy. However, distributing storage introduces significant challenges regarding consistency and reliability; client-side caching can improve performance but risks creating stale data if a crash occurs before writes are committed, while network partitions inevitably force systems to trade off between availability and strong consistency.
To address these complexities, two major distributed file system architectures are examined: the Network File System (NFS) and the Andrew File System (AFS). NFS operates as a stateless protocol that avoids relying on server-side open/close sequences by embedding all necessary context within each request, utilizing either hard mounts which block indefinitely during outages or soft mounts which return errors immediately. While early versions of NFS used write-through caching to ensure consistency, later iterations adopted weak consistency models where clients periodically poll the server for updates, leading to bandwidth inefficiencies even when idle. AFS improves upon this by employing a callback mechanism that notifies clients directly when files change, eliminating unnecessary polling and allowing changes to remain local until a file is closed, at which point they are propagated globally. Despite these advancements, both systems face limitations regarding scalability; NFS struggles with large numbers of concurrent writers due to constant polling overhead, while AFS relies on a central server that acts as a single point of failure and performance bottleneck for recovery operations involving callback state management.
Beyond traditional file systems, the lecture introduces Key-Value Stores (KVS) as a scalable alternative designed to handle petabytes of data across billions of keys using distributed hash tables. KVS simplifies storage by replacing complex hierarchical file operations with simple put and get interfaces, where human-readable names are hashed into uniform bit strings to ensure even distribution across nodes without clustering issues like name collisions. These systems must address diverse challenges including incremental node addition or removal, hardware heterogeneity, and load balancing, often utilizing either recursive directory architectures that create a central lookup bottleneck or iterative approaches where clients contact storage nodes directly. To maintain consistency in the face of frequent machine failures and network partitions, KVS implementations frequently employ quorum consensus protocols; by requiring writes to acknowledge $W$ replicas and reads to fetch from $R$ replicas such that their sum exceeds the total number of copies ($N$), systems guarantee that readers always see versions incorporating prior updates without blocking indefinitely.
Finally, achieving massive scalability in distributed storage requires sophisticated techniques like consistent hashing to manage data distribution efficiently as nodes join or leave the network. Instead of maintaining a massive directory table where every node must know about every other key location, consistent hashing arranges keys on a logical ring based on their hash values; each physical node is responsible for storing keys that fall between its own position and the next neighbor's position in this sequence. This design ensures that when a node fails or joins, only the data segments adjacent to it need to be moved rather than redistributing the entire dataset across all machines. Algorithms like Chord implement this concept using local knowledge of neighbors to handle dynamic changes with minimal overhead, enabling peer-to-peer systems and large-scale databases such as Amazon DynamoDB and Google BigTable to operate reliably at global scales while balancing consistency requirements against availability needs defined by the CAP theorem.
Read the full video transcript
okay welcome back everybody to uh the
last I guess official lecture before the
end of the term we're going to have
another one on Wednesday which is going
to be a special topics lecture but um
I'd like to continue where we left off
we were talking about uh distributed
storage and um if you remember before we
got into that topic we were talking
about the remote procedure call idea and
the idea behind a remote procedure call
is really that a client can link with a
library that includes a bunch of stubs
that allow it to essentially uh make
function calls which go all the way
across the network to a server machine
with the return coming back and they can
deal with them just as they would a
local function okay and so that's a
remote procedure call we're making a
procedure call remotely and some of the
key ideas we talked about were the fact
that uh the arguments to these
procedures have to be packaged up uh by
the client's dub and they're packaged up
in a network uh independent way and uh
serialized as a set set of bytes excuse
me as a set of bytes and then they're
sent across the network where they're
unpacked and the server stub will then
call a server function uh with the
deserialized versions of those arguments
and then the return call will get
serialized again sent across the network
and uh received and it'll be returned
into the client as a return from a
function call and so the client can
therefore use this regardless of the
fact that it's remote and um the couple
of things that we talked about were
among other things how these stubs get
generated uh there's a special IDL
language that you um use to describe the
procedure calls and a compiler that
generates the stubs for the client and
server side and uh you can basically
have the server be
remote of course or local and the client
doesn't have to know the difference
other than a um difference in
performance okay now today we're going
to actually show you an example of uh
use of RPC which is pretty common which
is to make a an actual remote um file
system work okay before I pass on from
this are there any questions
all right so then the other thing we
talked about is we talked about the cap
theorem the consistency availability
partition tolerance theorem which uh
really was more like a um a conjecture
by Eric Brewer back in the early 2000s
but uh it has since been proved in
various ways and the basic idea is that
you can have uh two out of these three
you can't have all three so you could
have consistency availability uh
partition do an you can pick two of any
of those three and um the thing to keep
in mind here is basically consistency
means that uh when you change the file
system on one side everybody sees those
changes
consistently availability means that you
always have the ability to access the
file system and partition tolerance says
that the network can uh survive being
cut in half
okay ah so um before I guess we have a a
late uh question here about RPC which is
which is fine the question here is if
the client sends pointers as arguments
does the client's tuub have to load all
of that in so pointers basically don't
mean anything cross machine so um part
of that serialization has to actually be
taking uh any data structures that are
consisting of pointers and serializing
them uh into a complete set of bites to
send across there are sometimes a
specialized uh ways of Packaging up
opaque pointer references and sending
them off to a server but the server
doesn't know what to do with them they
would only be for returning back later
to the client so I think the short
answer to the question is yeah if you
have any structures made out of pointers
they have to be serialized into bites
before they're sent across otherwise
they don't mean
anything okay so this cap theorem uh by
the way just to finish that uh thought
here will um have an impact on pretty
much any remote story that we might have
to deal with um and it certainly comes
into play when we start talking about
cach
consistency um of the the file system
okay all right are there any questions
on the cap
theorem all
right
so um so let's talk about distributed
file systems then so as you can see the
idea here behind this figure is really
the idea that the storage is going to be
in the network somewhere or we today we
call it the cloud I guess and you can
use that storage no matter where you are
you could be here at Berkeley uh on the
Left Coast you could be uh in Boston on
the right Coast or in Beijing whatever
and you can still use the data and in
some file systems you can even use the
data while you're driving uh from one
Coast to the other and that's all
because it's in the middle but once you
start having things remote in the middle
here then you start running into the cap
theorem so um what is a distributed file
system well it's pretty simple you've
all used this uh many times but we have
a laptop here and a server that's
actually got the data and so instead of
the file systems like we've been talking
about the last uh several weeks uh which
are local in this case you're actually
sending your uh request to read and the
response is coming back over the network
and the server is actually a separate
node somewhere else from the client
that's using
it okay um so a question that we might
have here uh which is in the chat is is
a solution to make the network resistant
to partitions so it doesn't have to be
partition tolerant so the problem with
that is pretty much that um we run up
against the end to end theorem which
really is how much work do you want to
put into the middle of the network to
make it so that it never partitions and
in practice uh you can add a lot of
redundancy to the network you can have
many alternate paths that uh can be
taken but uh ultimately it gets very
hard to prevent their uh never being a
partition in the network but you know
you can add a lot of redundancy so if
you don't take the path straight from uh
Berkeley to Boston going straight
through maybe you go by way of Alaska
and down if you have enough alternative
paths you can sometimes make the
probability of partitions uh very
low so what we really want with a
distributed file system is this idea of
transparent access to files on the
remote dis so that the client doesn't
have to know that this is remote from
the standpoint of the way you interact
with it the only know the only way you
notice is that things are slower okay
and so one of the things that we uh have
as a concept here is the notion of
mounting uh a remote file system onto
the local file system and so here's an
instance where um we actually have the
local root that's a little slash up here
and uh Slash users and then SL users
slash Jane we've actually mounted uh
another file system on the the server
called Kuby um and uh the partition Jane
is uh at this point in the mount and
then um inside that Jade file system
there's another uh directory called
program Prague and that we've mounted a
different partition from kubby to Prague
okay and so what happens there is that
the laptop user says SL users
jprog F.C in real ity because of the way
we've mounted this it's really in the
slrg uh partition on the kubby file
system and uh it's the file F.C and so
by mounting we can essentially get
transparency against these the fact that
these are actually remote so the local
user doesn't have to know the difference
so that's a form of transparency that we
get with the mount system
call okay now of course that raises all
sorts of questions which uh we don't
have a lot of time left in the term to
answer uh but uh one naming Choice which
you see pretty clearly here in this
figure on the right is that every file
in principle is a tuple of a host name
and local name in the file system and uh
we basically everywhere Below in the
operating system we always talk about
files as a tuple of host name and local
name so this is the simplest thing to do
and it's what we often do uh for
instance in the the department Etc uh
it's fine except that it doesn't give
you a lot of opportunity to move files
around to load balance or to try to deal
with failures it does let you do DNS
remapping so if kubby the file server
went down uh I could change its IP
address to point to a different um
server and then I'd still be up and
working uh another alternative though
which is much more interesting uh in the
grand scheme of things might be a global
name space where every file name is
somehow unique in the world and there
have been several instances of that over
over time today we'll talk a little bit
about one which can be uh based on
hashes over the
name
okay so let's talk about what's involved
in making a remote file system work so
we we've talked a lot over several
lectures about how to make local file
systems work but what about remote ones
okay so somehow the device driver I'm
going to put that in air quotes here
talking to the dis is got the network
involved so that's a little bit strange
already right because we think of device
drivers is going uh from the operating
system down into a controller and to the
local dis but instead we're going from
the files we're going from the system
call interface into the network and then
over to a different server and then
going into the device driver and so we
need some abstractions to let us do that
and so one of the abst ractions is one
called VFS okay so this is used it was
originally the virtual file system and U
then in Linux it became the virtual file
system switch I'll show you why switch
kind of makes more sense maybe but um if
you take a look at what I've circled
here in our kernel uh the file systems
actually go through a layer uh of
handling files and directories uh which
is called the VFS right there and Below
VFS is potentially any file system types
some of which are over the network okay
and so some of these file system types
might actually then interact with the
network subsystem go out come back
through a different network subsystem on
the other side and then back into the
file system and down to the block
devices okay so this VFS is going to be
an enabling abstraction that's going to
allow us to mount file systems first of
all of many types and then second of all
including things that are across the
network okay so um what exactly are we
talking about here so if you remember in
our layers of IO we talked about you do
a read uh system call it takes you into
the kernel um read which uh or into the
um lib C version of read which does a
system call takes us into the system
call processing and then if you look
down here this is by the way a slide
from lecture 10 or whatever if you look
inside we actually have something called
VFS read which gets called uh from the
higher layers and ultimately from the
user VFS being virtual file system okay
and so inside that call is going to be
interacting with the VFS layer and this
VFS layer you can kind of think of this
way so you got the the client process at
top comes through the VFS layer and
depending on which part of the uh
directory we happen to go to remember
the the mounting we could be going into
a EXT2 or three file system um kind of
like bstd or we could go to MS DOS fat
file system
and either of those could be used in the
same way by the client okay so this idea
of you know uh opening SL floppy test um
and then writing to sltm test what I'm
actually doing in this Loop here is I'm
reading from an MS DOS file system
writing to a Unix file system and this
all works because of the abstraction of
VFS okay so that's pretty good right so
how does that work so the VFS layer is
is like a local file system without any
of the diss involved and it's really
just a set of hooks that allow you to
plug in
functionality uh that's needed for the
client to act with a file system okay
and it's compi compatible with all sorts
of local and remote file systems and it
basically allows of the same system call
interface above uh regardless of the
file
system now we won't go in this in great
detail but for instance you could you
could look up VFS in Linux and it would
tell you that this is a an interface um
with four primary objects uh there's a
super block object an iote object a
directory entry object and a file object
that represent all of these things that
we talked about pretty much when we
talked about Unix file systems what's
interesting about this though is what
depending on what file system you plug
in it may not even have an inode object
think about the fat file system right
there's no iode there so really um what
happens is this VF s layer gives the
underlying connector the ability to fake
something that looks like a Unix file
systems you can make uh it look like
directories are made out of files even
if they're not you can make it look like
they're iodes and super blocks and so on
um regardless of whether those things
are really in the underlying file system
and so that layer sometimes we call that
a shim layer basically allows you to
plug in things that then the VFS layer
can make look like file systems okay
so um I'm going to talk to you about uh
NFS which is in some sense the first
user of VFS back when it was the virtual
file system um and so but this has a
persisted you know to this day so it's
persisted for the
last 20 years 25 years
so so let's talk about a simple
distributed file system in a little more
detail here so first of all we talked
about RPC so we're going to be making
procedure calls um so the client when
they need to do a read what happens is
the read goes into the VFS layer the VFS
layer uh then could just go ahead and
make a remote procedure call to the
server that's then talks to the dis and
gives you the blocks back and Returns
the data and so we could have a whole
bunch of these uh round trips and
because this is RPC we could even not
care about the the endianness of the
client versus the server because
basically the client caller procedure on
the server and it just works okay and so
um this is kind of the first way that
people build file systems um you know
you're using the remote procedure calls
to translate things but there's no local
caching in the client just in the
server um so the advantage of this is
It's the server is providing a
consistent view of the file system like
it does now if you were running
processes on the server um so that's
good the downside here is it's really
not performant okay it's expensive to go
across the network even when you're in a
local um you know even when you're in
the local network where it's going to
cost you a millisecond to go round trip
and much worse if you happen to have to
go uh to the metropolitan area or
globally where you're talking 10
milliseconds 100 milliseconds that adds
up really quickly for every block read
okay and so this is fine from an
abstraction you know gee we could build
this throw it together really quickly
this is really not going to work well
okay and there are actually uh ways of
mounting a remote server with uh SSH for
instance that kind of act like this okay
where you just open a tunnel and you
essentially get a mounted file system
it's really not going to perform very
well okay but you can do it um so
obviously the thing to do is caching
right so that's we've talked Lots about
caching remember everything in an
operating system is a cache you can
quote Kuby on that if there's nothing
else you uh get out of this class you
can you can quote me on
that so what we're going to do is we're
going to put caches in the system at the
client side in addition to the server
side so the server side cach is kind of
easy because that's the buffer cache but
we would like to for instance use the
buffer cach on the client side and you
know how does that work okay so the
advantage of this is if you can somehow
do the open read write close portion
locally because maybe you cash
credentials and information about some
remote file this gets really fast right
so the very first read to um some file
you reaches out you get an RPC across
the network pulls it off the result off
the dis uh puts it into local cache
returns it gets in the uh excuse me puts
it in the server cache returns gets into
local cache and returns a result and so
that read was slow the first time but
boy these subsequent ones bo bo are very
fast right and theyve just return the
value that's in the cash so that sounds
good uh but what are some problems with
this right so one of them is failure so
consider this idea here we have a writer
on some different client they write some
data in the cache and poof that machine
crashes and notice what just happened we
just lost data and that's because the
data was cached on the client and never
made it to the server and uh it's now
you know gone to Dev null so clear the
moment we start putting caches into the
system we've got some data uh
reliability issues we have to worry
about and of course we could force
ourselves to do an RPC with an
acknowledgement back first and
then uh return from the client so the
client never gets back an okay until
they know the data has been uh placed on
the server so that seems like a simple
fix because now if we crash we haven't
actually lost the data right
what are some other problems well
something else that rears its ugly head
which you probably can see here you know
this first cache is got uh the first
value in it the second caches got the
second value and so if client one reads
they get V1 and if client two reads they
get V2 and we have a serious cash
consistency problem okay now um the
question in in the chat is uh frankly
the obvious one which is how the heck do
you deal with this right so this is um
uh so on this slide this seems like a a
problem right so this is a problem now
uh we're going to talk about some
solutions to this but you could you
could start imagining some of these like
whenever you write you have to first
invalidate uh all the other caches and
then you get write to write and so when
they go to read again they get the next
one back right or you could say well a
little bit of inconsistency is okay as
long as I pull to get consistent data
back all right right you could send yeah
you could have changes there are many
options here um the first uh you know
the way they say this is the uh the
first step is to recognize you've got a
problem okay and so um the other thing
by the way to keep uh that I'll point
out is if for every right you're always
broadcasting the results uh potentially
you're using network bandwidth and that
may or may not be the right thing to do
okay so what's good about uh the
questions you all are asking here is you
got the right point of view um this is
clearly an issue okay so let's talk
we'll talk a little bit about what you
can do okay but um before we get there
let's talk a little bit more about
dealing with failures so we kind of
talked about maybe if you acknowledge
all the rights you can uh you know save
your data or whatever but what if in
general the server crashes okay so in
that instance this is not a client
crashing this is a server and you might
say can the client just wait until the
server comes back and keep going um in
many cases uh the client can't wait that
long because who knows how long the
server is going to take to reboot um and
maybe changes that are in the server's
cache uh but not on disk get lost okay
so when we talked about for instance the
buffer cach holding uncommitted results
there is that window of time where the
server might crash and the data isn't
there so clearly we want to start with
good journaling on the server side so
that's something we already know how to
do um but we need to uh yes so we'll
probably assume that the server is doing
its best uh to do some sort of Jour
journaling um raid whatever you take it
they're going to do it okay but um
something that's a little more subtle
here might be well what if they's shared
state so think about this for a moment
client doesn't open okay you by now
you're you're experts in using the open
system call and then um it does a seek
so it says well start me out at by
number
50003 and then it's going to do a read
okay now the issue with that sequence is
on the local Ser on the local file
system that just works right because you
seek to bite 50003 and your next read
starts there but in the case of a remote
file system if you do that seek across
the network and then the server crashes
and then comes back up or something and
now you do your read probably the wrong
things going to happen Okay so this idea
of shared state where in this case we're
sharing the state of our current uh
pointer in the file between the client
and the server that actually leads
itself to some really weird failure
modes
okay a similar problem might be this
what if the client goes ahead and
removes the file but the server crashes
before acknowledging maybe the client
doesn't know whether the file was
removed or not okay and if removal was
part of a cleanup process or um who
knows uh maybe it was part of a
temporary build there could be all sorts
of weird things that might happen
because that file is actually still
around even though the client thought it
was
deleted so one thing we can do is to
change our thinking a little bit and try
to make sure that uh all of our
interactions with the server are
stateless so a stateless protocol is
basically one where all the information
that we might need to service a request
is included with the request okay and so
you're um all very familiar with this
idea behind HTTP because when you go to
a website uh typically the uh state of
your access is kept in cookies on the
client side and all of those the
important cookies get sent with every
request and so as a result um the server
doesn't have to hold on to any
information okay um so maybe in the case
of a file system Maybe what we do is
instead of actually setting the um
setting the pointer to where we are by
seeking maybe what we do is we pass
under the covers this idea of uh well
I'm at 503 please give me some btes back
okay and that would be a stateless
protocol okay now um the question here
might be could would an alternative be
that we could make a bunch of requests
and then wait for a bunch of Acts um and
yes we could do that but once again
we're starting to get into that uh weird
uh generals Paradox kind of position and
um if we can just do everything
stateless it's much simpler uh we don't
have to worry about what the server
knows and what they don't so an even
better adjunct to stateless protocol is
empit and operations which say that not
only is the protocol stateless but if I
do the same operation multiple times
it's okay um because it'll just uh
ignore the next couple of times so that
would be the difference between a write
that uh file system that appends and
then I have to be sure I append only
once or uh a write to a block in the
file or block on disk which I can do as
many times as I want and it'll
eventually you know the data will get
written and if I it's already been
written and I write it again it won't
change anything so that's an idempotent
operation okay um timeouts that happen
to expire without a reply and with an
ident depit an operation you can just
retry okay so the the idea of stateless
is very appealing um for many reasons
like this um and again HTTP is a good
example of a stateless
protocol so the question uh might be can
we make use of that and we will all
right I'll tell you about NFS which is a
stateless protocol by
Design okay I want to do a couple of
administrivia things before we get
there so our last
midterm um and again there's no Final in
this class keep that in mind is this
Thursday 500 to 7:00 p.m. all material
up to today is included there um
although we'll be focusing on the last
third class we're assuming you don't
necessarily forget everything from the
beginning of the class um we're going to
assume that cameras and zoom screen
sharing are in place and um you know
there's no excuse to not have this
turned on so um you can lose points for
not having the camera and screen sharing
turned on um when the TA come talk to
you about it uh we might remind you at
the beginning but you it's really on
your it's on you to make sure that
that's all working okay and we're going
to once again um distribute links like
we did last time because I think that
worked pretty
well um there's G to be a review session
tomorrow from 7 to 9: um I didn't look
tonight but um I know that there is
already a zoom link for this that's been
uh put together and so it should be
published on Patza just watch for that
um lecture 26 which is uh Wednesday
won't be on the exam but it's going to
be a fun lecture of topics uh of your
choosing should you send them to me um
and uh so feel free to send me a couple
of
queries and I will do what I can to get
that in the lecture um baring that I
have some other things I'll talk about
I'll talk a little bit about data
capsules which I'm working on Etc
okay you can let me know just by uh send
me email okay um that seems simplest
right
now okay uh oh the other thing is um as
with last term hkn is virtual uh I'm
gonna I was going to post a video on how
to make sure you get a chance to comment
on the class uh we'll post that up on
Piaza but um don't forget to do your hkn
evaluations that's always
useful and uh I think that's all the
administra had unless anybody else had
other
questions okay I'm sure you all do very
well on the midterm um I'm offering you
uh good wishes for that uh in
advance and I'm going to miss having our
little time here uh every night in uh
Pacific time I don't know whatever time
it is for you guys some of you are uh in
very vastly different time zones I know
know
okay so let's talk about the N the
network file system from Sun Micro
Systems um this was uh in the 80s this
particular file system came
out and uh was in pretty pretty
widespread use uh still is in wide use
there's three layers for this which
you've already are now aware of the
first two there's the Unix file system
layer which is the uh the system call
layer open write close U file
descriptors pointing at file
descriptions you're all very aware of
that the VFS layer is this layer I just
introduced you to which distinguishes
local from remote files purely by uh
plugging in a table full of uh function
functions that are called as a result of
the uh system calls and then there's an
NS NFS service layer which is the bottom
layer that's the part that handles the
NFS protocol um does the RPC translates
and serializes into a network
independent form
format okay um the NFS protocol uh has
uh xdr is the serialization protocol for
for that RPC it was one of the first
ones out there um in fact uh NFS may
have been one of the very first ones to
have a network independent RPC layer um
it uh has uh operations for reading and
searching of directories manipulating
links accessing file attributes Etc are
all part of that protocol and that's
across the network in the NFS service
layer the other thing that it has um
it's certainly the first version NFS
version 1.0 and
2.0 uh had this uh very visibly shown to
the reader or the reader the user is
write through caching which is modified
data is committed to the server's disc
before results are returned to the
client um that got relaxed a little bit
over the years where um the client might
return while this caching still going
through from the buffer cach layer at
the client but um by and large it's a
it's a right through approach where the
transactions aren't done until it's
committed at the server side and so this
can slow things down quite a bit under
um various circumstances but you have
that advantage of knowing that your data
made it
um uh servers are stateless so the
protocol is a stateless protocol as we
were discussing so reads include all the
information for all the operations for
instance so you know when you say you do
read at uh I number position not read
file name okay and that's so that we
have all the information like for
instance the current position I want to
read at is included in the protocol okay
and there really is no need to uh do an
open close on the file across the
network because uh the local client has
enough information um and every
operation's uh satisfied on its own
okay all of the operations are
idempotent as I mentioned so you can
perform requests multiple times and it
gives you the same effect so examples
are the server crashes between a disio
and message send the client just resends
it the server does it again so that's
fine um you read and write file blocks
you just reread or rewrite and there's
no other side effects um the interesting
one here what about remove so if you ask
to remove a file from a directory and if
s uh may do the operation twice if uh
there wasn't an acknowledgement for some
reason the second time there's just an
advisory error that's returned back from
the server saying well that file wasn't
really there um so this is the kind of
adaptations the protocol to keep it
stateless and ident
okay the failure model for NFS is an
interesting one as well it's also
transparent to the client system uh in
general so the idea originally was that
when a server fails the client just
freezes until the server comes back up
and it just works okay and that was
called a hard Mount uh the problem with
that is that servers would go down and
then they would have all these processes
that were Reading Writing files uh from
an NFS partition and what would happen
is they would all get stuck in the
device driver and if you try to do a PSA
and see what's going on in the processes
is you'd see all these processes that
were all blocked with a little D and
that was that they were hard locked in
the NFS driver waiting for the server to
come back up and what's worse is that
was a an unkillable state so you
couldn't even kill them off they were
just really jammed up so that's
transparent but you might argue whether
or not that's a good thing okay and
there was actually a different type of
NFS Mount which uh is what everybody
pretty much uses today which is called a
soft Mount And if you do uh if you do
some um man on the NFS clients uh and so
on or do some Googling on that you'll
see about soft mounts the idea in a soft
Mount is that when the server goes down
you actually just get an error that
comes back um and your read or write
operation you were trying to do just
fails now of course that failure is kind
of weird because the client wasn't
expecting it to fail by a server
crashing because you're using the same
interface you would with the local file
system but at least it's not locked in a
way that can't be killed
okay so here's a picture of the
architecture as I mentioned so on the
client side we have the system call
interface uh which takes you through VFS
and then VFS has a whole bunch of
different possible file systems that
might be plugged in how do you know
which one to go to well depending on
what you mounted we showed you mount
earlier um the part of the file system
you happen to be in tells you which of
these
branches which of these actual file
systems you're going to use okay and if
it happens to be a local one you'll Lo
use the local file system if it happens
to be a remote moded NFS file system
you'll come off of VFS into the client
NFS client software which will take you
down into the RPC xdr layer which will
go across the network come back up into
the NFS server layer which uh comes up
into VFS which then um or or uses VFS to
access the local file system okay and
then the results get reversed in the
back the other
direction okay
questions so if you notice at the remote
side with NFS at least you're actually
just using um a file system on the other
side so um positive thing about this
idea here is that if the server is
disconnected from clients you can go
through and evaluate the um consistency
of the file system and so on with all
the normal tools because it just is a
local file system to the server and then
once it's operating as an NFS uh server
which You' get by starting up the NFS
demons then um remote clients are able
to access that file system on the server
that
way
okay so that's pretty cool right works
pretty well um but let's talk a little
bit about consistency of the caches so
the NFS protocol is a weak consistent
protocol by its nature so the client
actually pulls the server periodically
to check for changes um and if the data
hasn't been checked in the last 30 or 30
seconds 3 to 30 seconds it's setable to
some extent um then it plls and asks the
server what's the state of this
particular block and when a file is
changed on one client the server is
notified but that isn't reflected back
on other clients that happen to be
caching it it's up to them to to pull
and pull the changes Okay so in this
scenario that we had earlier where this
second client writes um and you get an
acknowledge back we can actually
acknowledgement we can actually be in a
situation where these two clients or at
least over the short term are um
inconsistent with each other but because
of the way this polling Works eventually
uh this first client will get the new
data okay so that's why we call it a
weekly consistent uh weekly consistent
protocol because the client kind of
conver merges to the right contents of
the
cache so for instance is F1 still okay
no here's a new value and at that point
the client is good to go with the latest
data so now so um there have been
various changes over the years that have
made it less likely to notice this
inconsistency clearly you don't want to
be polling so frequently that you're
using up a bunch of network bandwidth
and in fact the polling uh is a hard
limit even regular simple polling not
too frequently is a hard limit on the
number of clients that can be connected
to a server because every poll that
comes in from a client is using up
bandwidth on the server um and so you
know NFS clients can only be uh limited
number of them connected to a given
server but um if multiple clients write
there is the there are these windows
where things are a little bit out of uh
consistent consistency inconsistent um
and uh it is interesting you know when I
first started using NFS many years years
ago um I did notice that uh you I would
edit on one machine and I'd compile on
another one and occasionally I'd save
out some changes to a file and I would
be so quick at going to compile um in a
window to a different machine that I
would occasionally get these really
weird Phantom errors which were because
sort of part of my C file I had saved
out uh was intermixed with old versions
of it uh because of the NFS consistency
now this thing about why Google uh can't
handle hundreds of users
simultaneously um is
some is not quite the same issue here
because that there there's polling that
goes on and so at that point um you do
have to worry
about so there isn't pulling there's
actual pushing of data going on in that
case if you change too many things and
there are too many clients then you're
you're using up bandwidth going the
other direction the problem with NFS is
that even if nobody's changing anything
you're polling all the time and that's
using up bandwidth just while you're
idle so at least in the Google case
you're you're using up bandwidth only
when they're actual changes going on
okay now um let's but let's explore this
weak consistency for a little bit um
because what sort of cash coherence
might you expect from a system uh
if you didn't know it was weekly
consistent uh so suppose uh we have
three clients and client we start with
file contents um has a in it let's just
say and client one is reading at the
very beginning by the way time is uh
left to right here so um client one
starts reading and they're going to get
a um and client two starts reading well
they're going to get a for part of the
time but then if client one writes B at
some point client two might start seeing
B so there might be some intermixing of
b or a and then client two might write C
and you can get the situation where um
transiently at least you're seeing parts
of each file okay and um so what would
you actually want well one thing you
might want is what if I want to have the
same behavior as I would on a local file
system okay and if you wanted that um so
we have three processes instead of three
clients then you might want to say if a
read finishes before a WR starts you
always get the old copy if a read starts
after the right finishes you always get
the new copy and otherwise you get
either copy and it turns out that this
NFS polling protocol doesn't quite give
you that semantic it gives you this
little bit less clean
intermixing
okay all right now I'm seeing some good
uh combinations uh in the chat here
thinking of different op options between
polling and pushing um I'm giving you
that I'm going to give you the pushing
option in a section second here and we
can um ask some questions after that um
so for NFS rather than this somewhat
cleaner um view that we might expect
from a local file system we really have
this other idea where if a read starts
more than 30 seconds or pick your
polling time interval after a write you
get the new copy otherwise you could get
a partial update um so that's
more bandwidth efficient than it might
be if we tried to make sure that every
update was propagated to every client
all the time
but it uh it does have that slightly
weird semantic
okay so the pros and cons of NFS is it's
simple relatively so it's highly
portable so they were one of the first
ones to have the RPC with a
serialization xdr protocol some cons
though is it's sometimes inconsistent in
ways you can see and it doesn't scale
very well to large number of clients
because even in the idle case
everybody's polling okay so let me tell
you about another uh file system in this
space so this one came later than NFS
but not too much later so um I remember
working with the Andro file system AFS
in the uh late 80s and
um it became actually uh the DFS system
IBM bought the file system at one point
it was a commercial product um
it had a call back mechanism instead of
the polling so the idea is that this is
no longer stateless by the way so we're
we're um removing the ability to be
stateless but the server keeps track of
every uh machine that has a copy of a
file and whenever there's a change the
server tells everybody with an old copy
to invalidate their copy and uh as a
result there's no polling bandwidth okay
there's just invalidation bandwidth now
notice the decision that was made here
is not to push the changes out to
everybody who needs them or who is using
them but rather to invalidate and
there's another interesting option here
which is uh not option an interesting
semantic which AFS did which is
basically what I call right through on
close so think about this a second
Andrew file system AFS was really
designed to work in a much more global
environment than NFS um in fact you
could mount file systems uh that were
served in other parts of the country you
could actually Mount them and use them
locally and the performance was pretty
good and the reason for that is this
right through unclose consistency which
meant that when I open a file and I
start modifying it um none of my changes
are propagated to anybody until I
actually close the file even though I'm
doing rights it's not until I do close
and at that point my consistent ver
version is now available for viewing by
everybody else who's sharing the file
okay and at that point also that's the
point at which the notification goes out
that um there's a new version of the
file now in order to make this work uh
there are two things to worry about um
one is that um if I am have a file open
and somebody else changes it um I don't
want it to be pulled out from under me
so um what happens there is when I open
I
actually see the version of the file
from the moment I open it no matter what
else anybody else is doing okay so I
open the file they can be changing it
like crazy but I will continue to see a
snapshot of the file from the point I
opened it until I close it and reopen it
again okay so the upside of that is a
very consistent view I always have a
consistent snapshot of the file at the
moment I opened it and when I write
everybody always sees a consistent view
of the written product so I know there's
somebody worrying about race conditions
in the chat we'll get to that in a
second but if you think about that
relative to this NFS version AFS gives
you a much better set of semantics
because you never see an inconsistent
set of bytes in a file it's always a
fully consistent set of bytes okay so
that's that's an extremely positive
thing and um when we notify others that
the file has changed um they're either
going to keep working with their
consistent version or if they have it
closed right now they'll get notified to
throw their copy out and get the new
copy and they'll see a completely new
consistent version of what I've
got okay now
um couple of things here so if you have
lots of people writing they may not
actually see each other's rights so um
out of band you need a locking scheme or
a notification scheme to say hey I'm
working on the file right now why don't
you wait for me okay so that's uh one
thing you might worry
about okay the second thing that's
interesting about the Andro file system
is rather than caching in
memory okay which is what NFS does in
the buffer cache and your file system
actually caches on dis so the local dis
becomes a cache on the file system so I
can store whole
files um whenever I open a file the
whole file is allowed to be brought from
the server and put in my local dis and
now I can access it as fast as I would
if it were local because it really is
local okay so um so the potential here
is for a much better cashing because I'm
using the local disc to cache and so you
can have many many many more clients
talking to a given server because the
server isn't supporting every read and
write what it's doing is it's helping
with consistency
management okay
okay all right now
um now there's a couple there many
questions here I think the the way to
handle these questions is just to Think
Through what we've got here right so
when you open a file you get a snapshot
of the file at the time you open it and
you'll hold on to that snapshot until
you close okay and if you if you want to
do the equivalent of seeing whether
anything has changed you can close it
and reopen it you'll find out okay and
um if things never change or they they
don't change for a long period of time
because they're mostly readon then as
you use files they migrate to your local
file system and now you got really fast
action because now you open close read
read read do a bunch of stuff close all
of that is done purely locally because
the S the uh file server is responsible
for making sure that your locally cached
copies of the files um go away if
they're no longer
consistent
okay and if you just want to read a
small now good question what if you only
want to read a small part about it of
this file and it's a 20 gigabyte file
okay I think that's the question that's
being asked and that's a really good one
so the original version of AFS actually
you had to cach the whole file uh in the
local file system later versions
actually started caching in like 64k
chunks or whatever and so there was a um
there were modifications that allowed
you to have part of a file if the only
thing you wanted to do was read a little
bit of it and that took care of this
performance problem that you're worried
about here okay now um although I don't
talk a lot about this um okay I just
said that yeah I said this hold on I'll
say my other point in a second so data
is cached on the local disk um as well
as in memories um and on a right
followed by a close you send a copy to
the server which tells all the clients
with copies uh to invalidate their local
versions and that they'll need to fetch
a new version from the server at that
point um if the loser now if the uh if
the server crashes unlike with NFS we
can't even conceive of a client
transparent version of the protocol
because the server is supposed to have
all this callback state to keep track of
who's got copies of things so when the
server crashes it comes back up it
actually has to request uh information
from all of the clients that are
connected as to what copies of what
files they've got and so that's a little
that's more expensive for rebooting this
server okay so the pros uh relative to
NFS much less server load dis is a cache
so then technically the cache is much
larger um the callbacks means the server
doesn't have to be involved if files
read only okay and so you can if you
have mostly or totally read only
partitions you can have a small server
basically share a huge set of clients
because really all it's doing is helping
the clients get copies of the data onto
their local cache okay now for both AFS
and F NFS although AFS is less
problematic here the central server
becomes a bottleneck um and so the
performance of all the rights ultimately
uh go through the server um and so uh
there is a question about availability
because the server becomes a single
point of failure um and the servers has
to be more powerful than the clients and
so it's typically a higher cost than a
simple
workstation okay um now uh a good
question is brought up here which is uh
couldn't the server store um call back
State on the dis and the answer is yes
uh it probably has in fact as I recall
it has a cache of what it used to know
the server state was but who knows what
happened uh when it crashed and came
back up so it has to at minimum validate
what the current state of the um of the
caches
are all
right good so um one thing that's fun
about the Andrew file system which I
didn't write down is the Andrew file
system had the notion of global names
okay and so um if you were to look at a
client machine you would see that there
was a slash AFS slash partition and then
you could Mount pretty much anything
from anywhere in the world in a way that
was
independent uh was an independent name
and as a result um in principle every
file in AFS was uh globally available if
you had the right permissions um and so
this is a little different than NFS
where things are named by that tupal
that I mentioned earlier which is a
particular machine and a local file name
here in principal at least there was
Global file names or at least it was
starting to go that direction um and so
you would Mount you know we would be um
at MIT and we would Mount files that
were down at uh um CMU and ones that
were over at Berkeley and so on we could
Mount files that were uh on across the
country and it actually worked pretty
well because most of the performance was
handled by the local dis and so this is
an example of something where you really
were starting to mount things very
distantly okay and now of course you're
all used to that with the cloud but um
this was quite the Innovation back when
it first came
out all right but let's move even
further away and sort of ask uh you know
what's this Obsession that we have with
files uh what about sharing data instead
of files and one thing that's become
very popular over the last decade
actually I would say last 15 years is
this notion of a key Value
Store uh where the world is like a big
hashtable that uh lets us look up keys
and get values
back okay and really back in the early
2000s um when I started working on uh
peer-to-peer storage systems key value
storage were kind of in their early days
okay so really this idea has been around
for um you know more than 20 years it's
just that it's become very prevalent
over the last decade and it's native um
you know pretty much in any programming
language you got associative arays and
Pearl and dictionaries and Python and
maps and go and you pick your language
there's a there's a hash table the key
value store that we're going to be
talking about is kind of like a hash
table that spans the globe or spans the
network and so um you know for
everything you can imagine using a
hashtable for in these languages that
you're aware of you can use a key value
store for um more globally okay and
um in terms of sharing information what
about a collaborative key Value Store um
rather than message passing or file
sharing so rather than thinking about
taking file system mounting the file
system on two clients and then sharing
through files maybe we have a key value
store and we just happen to know what
the keys are that we're using and we
share that way that seems like another
option here and maybe we can have more
uh more options on how to make things
consistent and how to make them
durable so we might ask ourselves could
we make it scalable can we handle
billions or trillions of keys can we
make it reliable uh even though things
are failing in the Network's
partitioning and so on uh can we always
get at our data now um I will tell you
up front here we're not going to violate
the cap
theorem but what we can do is we can um
perhaps we can get to where the cap
theorem doesn't bother us quite as much
so we get an old value of the key that's
that's pretty close to recent maybe not
the most recent one and maybe that's
okay okay so the basic idea behind a key
value store is a very simple interface
okay there's put and get okay okay put
has a key in a value and what it does is
it inserts uh that value at that key
into the key Value Store whatever that
means you know it's it goes off into
cyberspace somehow get takes the key and
Returns the value from cyberspace
somehow okay so the interface is uh
almost boringly simple the question is
can we do something interesting with
this that uh
is scalable fall tolerant reliable
durable put your all of your favorite
ibles in there can we make that happen
out of this simple interface and the
answer is this becomes much the answer
is yes this becomes much simpler because
the interface is so
simple okay so why key values Thor okay
I've already said this but it's easy to
scale huge volumes of data petabytes
okay exabytes you pick your number um
big right you form items you can
distribute easily and roughly across
many
machines so if I have 10 machines versus
100 machines versus a thousand machines
I can just scale up the number of key
Valu um pairs I can handle and how many
clients I can handle just by adding more
things to the system okay and so that's
that's kind of appealing uh if you think
about a big NFS file server or a big AFS
file server or whatever your favorite
thing is
um the way you typically scale something
like that up is you go and you buy a
huge piece of Hardware okay and that
really big thing is fast because it's
got a lot of really fast processors in a
single box and it's really expensive uh
on the other hand the way you might
scale up a key value store is you just
to add more and more machines to it and
just this incremental scalability gives
you more power and so that's going to be
another um appeal of this idea okay
okay so properties are pretty simple
from a consistency standpoint because
all we want is well we can talk about
what types of consistency we might want
but one simple thing is perhaps we just
want to know what the latest value is
associated with a
key um and there are many cases these
days where this is a simpler but more
scalable version of a database um and it
you could think of it as a building
block uh for a more capable database if
you want better semantics than just uh
um you know what's the latest value on
something but um oftentimes a key
associated with a value is enough and
you can call that a database
okay good examples of this there are
many so Amazon um you know key might be
customer ID value might be profile
Facebook Twitter key might be the user
ID the value might be the user profile
um iCloud or iTunes the key might be a
movie or song name the value might be
movies or songs
so there are many examples of keys and
values that you use every day without
actually thinking about
it and by the way all of the big cloud
companies all have really good key value
stores that um scale really well and
people use all the
time so in this case um the good
question that's that's in uh the chat
there is so are Keys kind of the same as
Global file names in AFS yes roughly
speaking Okay so keys are um these
Global names that you could get at
anywhere in the system and if you had a
if you had a key value system that
spanned the globe and everybody was
using then the keys would be a global
naming scheme now thing that's a little
tricky about that is Keys uh if you if
you just have a key that say your
name uh the problem with that kind type
of key is it's very clustered right so
there are many people that have the
first name John uh and so there would be
a part of the key value space that's
really overused and then there' be lots
of places where it's underused and so
really what we use with keys when we
want to um really make this scalable is
we start taking names that humans use
and we hash them into a uniform set of
bits like 256 bits that is the global
name that these systems typically use
and it's a hash over the human readable
stuff so it's close okay it's a hash
over human readable
stuff okay but if you want to take the
simple uh version of that question about
our keys the same as uh Global file
names the simple answer is
yes now so in real life like uh like I
said here um Amazon has Dynamo DB which
is the key value store that's used to
power the shopping cart in uh amazon.com
there's a simple storage system or S3
which is uh key values storage that's
used for some of the big um cloud
storage services that people
use Google has big table hbase
hypertable several of these distributed
scalable data storage systems which
ultimately come out as key value stores
Cassandra is um was developed by
Facebook which but it's a key value
store that's used in a lot of cloud uh
processing there's mebc D which is an
in-memory key Value Store um that uh for
instance redis is an example of
something like MCD that then spans
multiple
sites edonkey emule there's lots of
peer-to-peer storage systems um before
any of these things uh we did research
in peer-to-peer back in the 2000s um and
so cord which I'll tell you a little bit
about toward the end of the lecture here
um tapestry uh was one that we worked on
cord was an MIT Berkeley version um
there was um a number of other ones that
are out there so these are all key value
systems that partic work particularly
well uh across the globe so all right so
the reason I brought this up is I just
want you to know that some of the ideas
we're going to talk about in the last 20
minutes here are basically used quite
widely
today now um the question here let's see
does files in key value store have a
smaller file size requirement than
AFS uh I'm not entirely sure what the
question here is there's nothing that um
sets the particular size of a value so
your key is the thing that might be um
Limited in being a
256bit hash of something the value is
oftentimes something that can be
anything from a small number of bytes to
gigabyte video or whatever um usually
the thing that's the limit is the the
maximum size not the minimum size I
don't know if that answered your
question or not
um okay so let's look at the basic idea
behind key value stores these are called
distributed hash tables often times too
so it's like a hash table but
distributed right so main idea is we're
going to simplify the storage interface
so we're going to get rid of all that
open close read write complication that
we Tau you at the beginning of the term
you know forget all that except for by
the way the midterm on Thursday and what
we're going to do instead is we're going
to do put and get and we're going to
partition it a set of keys and values
across many machines and so this thing
um
here this yellow key value huge table is
kind of the abstract space of all keys
and values and what we're going to do is
we're going to partition it across a set
of available machines out there so that
um you know each machine handles a range
of the space okay now I haven't told you
how to do that but the idea is in
principle that if you think of the space
of all possible keys and values actually
the space of all possible possible Keys
is really what we're talking about here
um you could easily say well here's all
the machines that I'm going to have
participate let's just distribute the
keys over them and uh make it work
somehow okay so that's going to be the
simple idea how do we make that work
well there's some challenges right so
one of them is uh whatever scheme we
come up with to do this mapping from the
abstract table to the physical locations
um we want to scale to thousands or
millions of machines
and so we need to make that index work
somehow right that's going to be a CH
Challenge and furthermore as I kind of
told you a little bit ago we want this
idea of what's often called incremental
scalability which is we want the ability
to add more machines as we need more
power and so whatever scheme we come up
with ought to be scalable in a way that
uh just
increases automatically or at least
easily um the other thing is there needs
to be some f tolerance here so when
machines fail because machines will fail
um we don't want to lose any data and um
one thing that we haven't talked a lot
about with failure U because we haven't
had a lot of time this term for this
topic but uh if you have a machine fails
once a
year um on average and then you put 365
of them together you're now going to
have a machine failure on average every
day okay because failures scale
uh inversely with the number of machines
so typical warehouses that Google and
Facebook and so on have which have
thousands tens of thousands of machines
in them have failures going on many
failures per day where machines are
coming up with some failure mode or
maybe their discs are just plain dying
or what have you but whatever scheme you
come up with needs to handle failure
very well because failure in this
instance is not an uncommon thing okay
just because of the scale um and then of
course consistency um is going to be
important so remember the cap theorem so
consistency says that basically
uh we have some way that many clients
that are writing all get to see the
readers get to see those values in some
consistent way or at least an eventually
consistent way where we we all agree on
what the latest version is eventually
okay and that consistency needs to work
even though there's failures happening
and you know Bas basically our cap
theorem says that maybe we can't stay
available consistent and um and uh
Network tolerance uh Network partition
tolerant all the time but it'd be nice
that when the thing that failed came
back we would eventually converge to
something so that's consistency all
right and um heterogeneity is one that
many times you probably wouldn't think
about if I hadn't put it on the slide
but the the issue here is really that as
I add machines over time these machines
are all from different
um
different purchases you know they're
from different purchases different Lots
um
different uh years different models and
so they're all going to be a little
different and so that means there
there's this huge het heterogeneous mess
of machines and network bandwidth and
latency and all of those things and
somehow we would like this system to
mostly work well despite that wide
ranging set of components okay so this
is a this is a large set of requirements
and you know nothing's going to be
perfect but we might want to have some
way of building our distributed hash
table so that we can handle at least
some of these things reasonably well
okay and that's going to be our
goal okay so some questions are for
instance if we do put uh key coma value
where do we store it well for that's
going to be complicated because we got
to start by knowing what's available and
if we keep adding machines and machines
keep failing then the wear might
actually be more complicated than you
might think and then of course when we
go to get there's a question of the wear
of where do we get it from especially if
machines are failing maybe that key has
moved around a bit since I put it in
there originally and so um whatever
scheme we come up with is got to handle
where very well right and then we got to
do the above well still keeping our
scalability and full tolerance and
consistency and all those other things
that we talked about
earlier so how do we solve where well
one way is we can take the key space and
hash it to
location all right and and so you know
basically if we knew these 100 nodes are
definitely going to be used used we
could build a partitioning that sort of
partitions from the the key itself to
one of those 100 places and you know
that might do the trick for us as long
as everybody knows the the hash key um
but you know what if you don't know all
the nodes that are participating or
maybe they come and go or what's worse I
mentioned this earlier um maybe if some
keys are really popular then you might
have machines in a in a partitioning
that was uh
you know partitioned equally among the
key space maybe some machines will fill
up whereas other ones will be empty okay
so the the wear we have to be careful
about trying to keep load balance in
addition to all these other things um
and then look up well if we if we build
this thing by having a huge table on one
machine that knows where everything is
uh that's going to be a bottleneck and a
single point of failure so um I hope you
guys can realize that uh at the face of
it we certainly are not going to do this
which is take this thing that I've shown
you here as a big table and put it on
some huge database server and use that
to look things up okay that we call that
the directory approach and I'm going to
show you abstractly what that means in a
moment but that would clearly not be
scalable or fa tolerant
okay now before I go a little further I
want to pause for a second and see if we
have any questions
okay
so let's look at a recursive directory
architecture or uh for put so let's
assume for a moment that this directory
is a thing it's on a a machine um
somewhere and we'll we'll fix that in a
in a few slides but um then way we would
do this is if we want to put a new value
for key4 we' go to the directory the
directory would say oh um I'm going to
assign key4 to node 3 um it would go to
node three and do the put we would get
an acknowledgement that came back um
potentially or or not but anyway what
happens here is the put gets redirected
through the directory to the storage
server we we're going to call this
recursive because what happens is the
put goes to the directory which goes to
the file server so it's recursively
going from one point to another um the
alternative is what we might call
iterative and I'll show you that in a
moment but how does the recursive get
look like well we go get to the master
directory it it knows where to go it
gets the value which comes back and the
directory forwards it on to me and so
this again is the get goes to the
directory which goes to the node and the
node goes back to the directory goes
back to the client um another way to
think of recursive uh structure here is
it's like routing we're kind of routing
through the directory here the
alternative is often called iterative
and in the iterative case which is
basically what's happening is we um the
client says I'd like to put key 14 the
directory says oh I'm going to put that
on node three use node three and then
the client says oh okay um node three
please put for me so notice that this is
iterative so the first thing I do is I
find the location that's and then I go
and I do the storage so I'm iterative
working through a set of locations in
the network okay um and then get
iteratively I get back to where the
location is and then I can go to that
server and talk to it
okay so um just putting them both on a
slide here we sort of have iterative
versus recursive so the or recursive
versus iterative I should really change
that title so the recursive case um is
potentially faster because we're routing
through the directory server and back
it's a lot easier for consistency
because we can make sure we know
everybody who's trying to change that
given location at any time whereas on
the iterative side we've got everybody's
kind of doing their own thing and
they're talking to the storage servers
independently of one another um the
downside of recursive is this directory
is definitely a performance bottleneck
the downside of the uh iterative is it's
much harder to enforce consistency so
they have pros and
cons so is it easy to make the system
bigger well we can add more
nodes
um and now maybe we can handle more
requests so we can serve request from
all the nodes that have a value in
parallel the master we could try
replicating and somehow um use it to
replicate uh popular items okay except
the master itself is going to be really
hard to make scalable uh we could try
making many copies of it uh but then we
got to keep them all consistent with
each other we could try to Partition it
so different keys are served by
different directories but how do we do
this and so while the the version that
I've shown you so far where the
directory is a thing um it seems like
it's uh definitely going to be an issue
from a performance standpoint and as is
pointed out in the chat here it's
definitely a single point of failure
okay um and it's it's really a single
point of failure for both the recursive
and the iterative versions because the
iterative has to start by asking the
directory where things are okay because
remember we're allowing things to move
around as failures happen
so let's uh let's talk about fall
tolerance in a couple of ways so one we
could replicate for instance the key on
many
nodes okay so that basically the um the
copy puts on several places so now uh we
never lose
data if uh if a node fails because we
have another copy if the master
directory fails then we lose
availability but in principle we could
scan through all of our uh nodes and
reconstruct the directory from the
actual data so from a fa tolerance
standpoint this particular scheme I'm
showing you here doesn't lose any data
okay but we still have this directory
being uh certainly an availability uh
single point of failure at
minimum okay but let's also talk about
consistency so we want to make sure the
value is replicated correctly so how do
we know the value has been replicated in
every node
and what happens if a node fails and
what happens if a node is slow um so if
we want to replicate 12
times and we want to make sure there are
12 cop copies then all of a sudden the
put becomes slow because put has to wait
for 12 copies and then it gets back and
act and then it can go
forward okay so in general if you have a
lot of replicas slow
puts uh are going to be part for the
course but potentially fast gets because
I could get get from any of the
copies
okay so let's look at a consistency
issue here so if you do put um uh right
now if you look down here k14 is stored
on node one and node 3 and notice that
it's v14 is our current version but now
some new client tries to put v14 Prime
and another client tries to put v14
double Prime and now if you notice uh
depending on network ordering we could
get a situation where uh the rights
between the directory and node one and
node three get reordered such that node
one thinks that v14 Prime is the the
most recent and node three SS thinks
that v14 double Prime is the most recent
and really if these two puts were
simultaneous there's not necessarily any
right answer as to what's the most
recent but we want to make sure that the
system has has picked one okay and so
the problem here is that get is kind of
undefined
okay so there's a large variety of
consistency models out there of what to
do when you have simultaneous rights
going on um there's linearizability
which is reads and writes get put to
replicas they appear as if there was a
single underlying replica so that's kind
of like transactions there's an ordering
um there's this eventual consistency
which I've been talking about where they
may temporarily be different but um some
anti-entropy process eventually makes
sure that uh everybody agrees on the
most recent copy
and there's many others okay and um
that's a different class but it would be
um you know I often talk about this when
I teach um 252 for instance uh the
architecture class graduate class
haven't done that in a few years but
there you start talking about causal
consistency and sequential consistency
and strong consistency and so on which
is really about what happens when
multiple people are writing multiple uh
different key values uh at the same time
how do you order all that Cor okay um
the simple one I want to talk about
today is called Quorum consensus and
we're going to improve put and get
operation performance in the presence of
replication doing the following so we're
going to say that put uh we're going to
say that there's a n
replicas okay and put's going to wait
for acknowledgements from at least
W okay before it goes forward and we're
going to assume that things are uh
timestamped in some way to make this
work um and the time stamp basically is
going to let um replic because let see
two things coming in it can replace uh
an older one with a newer one based on
the time stamp then um get is going to
wait for at least our replicas to say
here's a value and as long as W plus r
is greater than n what we know is that
if the put happened before the G then
the G will always get the most recent
value because the fact that W plus r is
greater than N means that any overlap
of uh is going to basically have at
least one uh replica that's got the most
recent copy okay so there's at least one
node that always has the update and so
this Quorum consensus is something
that's used pretty commonly now in
Cassandra and a lot of these other
systems used by Facebook used by other
cloud service providers um and it's up
to the client typically to pick WR and N
um but a typical value is that there's
three n is three you write to two of
them and you read from two of them and
as a result you'll make sure that you'll
always get the most recent
copy
okay now um I'll let this uh simmer in
your brain a little bit while you're
thinking through this but for instance
um the interesting thing about for
instance you could uh ask for uh three
you could have r equal to three which
really says that I go and I ask for
three copies and I wait till I get all
three of them before I decide or um I
could for instance get all three of them
and when one comes back or two of them
come back then I go forward that's
really what's going on so if I say r
equal two I'm really potentially asking
for all three and taking the first two
that come back and um that lets me
actually tolerate a slow server and this
w+r greater than n actually lets me
tolerate failures in that group of n so
this Quorum consensus has not not just
consistency uh positives to it but it
also handles failures that have happened
uh while you're writing between the
writes and the reads and so on and it
handles slow machines as well so qum
consensus is a remarkably simple idea
that has a lot of positive benefits okay
and the way you know what the updated
copy is is the timestamp so what I said
here is in red on this very slide um
basically when you write you're using
typically use a Tim stamp um that you
put in all the copies that go out there
and basically the um clients and the
servers are sorting by time
span okay and so responses um are
potentially returning not always the
same value but um they'll return R of
them and then you pick the uh the most
recent of those
are it's a fairly simple scheme but it's
fairly
powerful okay now um you might use w+r
greater than n plus1 for any number of
reasons including uh you know making
sure that you're really sure that you've
written um three copies for instance Etc
there there's a fault tolerance and
performance reasons for possibly having
R w+r greater than n plus
one um so here's an example for instance
um here's the initial put where we uh we
want to um we try to write to all three
of the uh copies but really we only get
acts back from two which is okay because
W equals two all right and so then later
when we go to read we read from two of
them and
um in this case we read from one uh but
we'll always get back uh the most recent
so we so even though we've read from two
of them one hasn't responded we always
get the most recent back because we got
that overlap uh between the um the two
that we've written and the two that
we're reading we know there's always one
overlapping one that will give us our
value okay and again the most recent not
the thing that's most recent is based on
that time stamp okay this thing that's
why I've got this in red
here okay so see the red everybody who's
wondering about most
recent
okay all right
now if you guys will hold on for just a
moment I I'd like to get a couple of
more things done here since this is our
last lecture um so storage uh the way we
get scalability is we want to use more
nodes we might have a number of requests
we can serve requests from all the nodes
that have the value stored in parallel
so that's potentially good the master
can replicate a popular value on more
nodes so we can get more performance
with more replicas uh in a scheme like
this um to to give us the master
directory scalability we could replicate
it we could partition it so different
keys are stored in different Master's
directories how do we partition it um if
I were you guys and this is the first
time I've heard this lecture I'd
probably think that Professor kual
hasn't really told me how to make the
master work okay because this I would
have this uncomfortable feeling that
yeah this sounds great but that Master
seems like a problem okay and so let's
see if we can do something so load
balancing um the trory keeps track of
the storage available at each node
preferentially insert new values on
nodes with more storage available okay
so I can see that might work when you
add a new node what you'd like to do is
you'd like to rebalance everything
somehow so that new node really starts
taking its fraction of the
load okay and so that sounds like
there's some rebalancing process that I
haven't told you about
here and then when a node fails we need
to make sure that let's suppose that n
was three and we were aning on the fact
that we had three copies of things if
one of those three copies fails we would
like to make sure that some other node
got a copy so we kept our basic
redundancy in there of three okay and so
that also I haven't told you how to do
that so let's uh as kind of our last
topic before we we really uh cut out
here is how how do we scale up our
directory so the challenge here is the
directory has a number of entries equal
to the number of key value tupal in the
system which could be billions or
trillions pick your pick your favorite
large number and so that directory thing
is big and really we want to distribute
it in the same way that we're
Distributing the actual data and the
solution here is something called
consistent hashing which hopefully um I
think you may have heard of in other
classes but it's going to give us a
mechanism to divide the key value pairs
amongst a large set of machines but do
so in a fully distributed way without
ever going through a single directory
machine okay and bear with me
but the idea is it's going to be simple
but it takes a moment to catch so the
idea is we're going to associate each
node um a unique ID in a a
ring uh of all the possible values from
zero to 2 to the m minus one and
typically m is going to be big it's
going to be the 256 bits in our hash
we're going to call that the ring and
then we're going to um partition that
space of possible Keys across n
machines and um all the key values are
going to be stored in a node with the
smallest ID larger than a key okay so
let me rather than um trying to catch
all that uh in words let me show you
this in a picture okay so here's an
example of the ring and what I've done
here is I've I've made
M6 okay so this is a six-bit hash space
really not interesting in the grand
scheme other than for class but if you
notice if m is six
then the set of all possible hash values
is from 0 to
63 okay and the idea that means this ID
space is from zero to 63 um and each
node is going to have a unique spot in
that space so node 8
um you know so this node eight I'm going
to put on a spot of the Ring node 15
node 20 how does a node know what its
name is well it's going to take uh
things like its IP address maybe the
name of who owns it and all that stuff
it's going to put it together into a
hash and it's going to Hash it to find
where its position on the ring is just
like the keys are hashes uh over data
that we want okay and the way we're
going to handle this is for any given
number of nodes which are hopefully
spread throughout the ring then the node
is going to handle every key from uh
just bigger than the previous node so in
this example node 8 is going to map um
b or um node 15 here is going to map
everything from 9 to 15 uh node 8 is
going to map everything from 5 to8 Etc
and that's going to store those keys
okay and if a node goes away then we're
going to make sure that the next node up
is going to store all the keys okay so
this is a very simple scheme for
consistently partitioning hash values
among the
ring okay so for instance the uh key4 is
uh is going to be stored on node 15 15
because uh node 15's the um the node
whose name is the first one clockwise in
the ring from the key I'm looking
for
okay
questions by the way um this thing I'm
talking about with consistent hashing um
does not is not going to be on the exam
we've talked about key value stores but
I want to um I I just wanted you guys to
see a real implementation here okay um
now the uh the different types of
machines we have a mixture of these
machines um spread throughout okay the
key thing to make this work and that's
no pun intended is that these be
distributed throughout the ring and the
way we get that is by having a good
hashing
function well you don't have to be aware
of all of the machines involved so
that's the part that's cool about this
which I'll have to continue on Wednesday
you guys may have to come back for this
but the cord algorithm is one which
adapts to nodes coming and going where
the only thing you know about is a local
number of the
nodes okay so if you look in practice m
is really 256 or
more
okay now cord is a distributed lookup
service that does this um the important
aspect of the design space is to couple
correctness from
efficiency okay and it's it's going to
um the correctness which uh goes along
with the question that was just asked is
uh that every node needs to know about
its neighbors on the ring and that's it
so if you go back here the only thing
that node 15 needs to know about is node
8 and node 20 and the rest of the
algorithm of cord basically uh takes
care of that okay and so um we're going
to talk about that on Wednesday um where
we've gone way past our time and cord is
not in scope for the midterm so that's
fine but I just wanted to leave you guys
with this uh interesting idea that we're
going to show you how to build a
distributed system we'll do that on on
Wednesday such that um we only need
local information which is about a few
nodes in the
system and a log number of other nodes
that are spread across this ring as long
as we know only that local information
we can do
highly efficient
lookup um and deal with failure as nodes
come and go and do replication in a way
that keeps everything um safe and so
that's going to be cord and we'll talk
more about that on Wednesday so I hope
you guys all come to that because um
court is one of my favorite simple
directory uh distributed directory
storage systems and so um please come
we'll talk about that we'll talk about
some other things but um I'm going to
bid Ado to everybody I hope you have a
great even evening um and good luck
studying for the exam and please come on
Wednesday because we'll finish talking
about cord on Wednesday um and we'll
talk about a few other topics if people
show up and um if I don't see you on
Wednesday um you've all been great and
I'm going to miss these little lectures
have a good evening and uh good luck on
the exam