In this video, we will separate input,
statistics and output logic into
separate threads.
Unlike other videos in this series, we
are not going to end up in a fully
functional State at the end of this
video.
However, we will end up with three child
threads that we spawn which do
everything that they need to except talk
to each other. So we'll be able to start
up the program see that it starts up and
shuts down in a coordinated fashion, but
it won't actually process our data. Let's
go ahead and do our imports, from the
sync module we need our canned mutex, arc
is our atomic reference counter, and
mutex will protect access to mutable
data. And then we need the thread module
so we can spawn threads. Since I need to
pass the three different arguments that
we parse to three different threads, I'm
going to go ahead and D structure the
struct into three local variables, that I
can move to each thread separately. The
other option would be to make three
clones of the args struct. So in file, out
file, and silent are three local variables
that I created out of args. Now we can
create quit which will be an atomically
reference counted, mutex guarded boolean
that starts as false, and this we will
use to coordinate the shutdown of our
threads, since we're not adding channel
communication in this video. Cloning an
arc just creates a new copy of it,
pointing to the same data, so I'm making
three clones so I can pass one to each
child thread, and still have one in the
main thread to use. The arc will
coordinate clean up once nobody's using
a reference to this data anymore, and the
mutex will protect our mutable access to
the data. Spawning a thread returns a
thread handle, we are going to spawn
three threads, so we'll have three thread
handles. First our read thread, will do
read handle here. The thread spawn
function takes a closure, and we will
usually use move closures because
otherwise borrowing
gets really really tricky. You can see
that I'm passing in my quit reference,
and I'm also passing in the local
variables that the functions need. You
can see that the closure doesn't take
any arguments because there's no
arguments between pipes. So we've got our
three handles here, read, stats and write
handles, they're really similar, and I'm
going to change these loop functions so
that they'll return the same result type
as well, right now the statuses returning
something different, so we'll fix that in
a minute. But right now we want to join
the threads, and crash our main thread if
any of our threads have crashed. Sounds a
little unintuitive but we're gonna do
some simplistic handling here. So when
you join a thread handle let's wait till
the thread returns, and when it returns
it will return a thread result that
wraps what we return from the functions
with just an I/O result. So we've got a
nested result here. So the first thing
I'm going to do is go through each of
these handles, and just call unwrap, so
the thread will really only return an
error if the thread crashes. So if the
thread crashes might as well crash the
whole program, so I'll just call unwrap
on each of the joins to get to the inner
I/O result, or crash. So here I've got my
read I/O result, my stats I/O result, in my
write I/O result. And now we're done
gathering all of those we can handle
them, let's look at each of them and
return an error if any of them are
errors. If they're success we'll just
ignore them, so we'll just use the
question mark operator to unwrap the success
and ignore it, if it's there or return
and error from main. Now if we take a
look at the rest of main, we see that
it's the main loop that we need to
refactor into loops in each of the
threads, will keep the okay at the end,
and in just a minute we'll come get this
top part and put it in the read loop,
which we're going to create write now. So
first we need to do our imports, which is
just our canned mutex from the sync
module, and then we need to go at our
quit parameter, which is our atomically
reference counted, mutex guarded boolean.
Since we won't be using return values to
communicate actual data, we're going to
change our results to just be the empty
results, just to communicate errors if
there's an error. The reader logic
doesn't change, but we do need to go down
to this return result and take the
buffer out of the okay. And then we can
go back to main and grab that block we
highlight it, and come back over to read,
and paste in the beginnings of this loop
right after we create the buffer. Let's
go down and grab the read call that we
used inside this function originally and
let's cut that out, and then we can go
and replace that call that we had
outside in main with the real read
function that gets the num read. And then
we can alter this match so that instead
of looking for an empty buffer, it looks
at the number read variable, which we can
match explicitly with a zero instead of
a guard, and the other number gets
returned directly an error gets broken
out and ignored, then we can clean this
up and close the loop, and add a reminder
that in a future video we'll need to
send this buffer to the stats thread.
Similarly, if we break out of the thread
we will need to send an empty buffer to
the stats thread to let it know that
we're done.
But for now we'll use quit to signal
that we're done. We'll lock the mutex and
then we'll unwrap the result that we
receive, it would only be an error if the
thread holding the lock crashed, and then
we can dereference and alter the actual
variable, and now we can go add an empty
tour okay, and then go up and remove this
total bytes line that was for stats, we
don't need that here. Oh and then we need
to add match on our match here, we missed
our match statement that should fix our
syntax, then we just do a little bit of
white space cleanup, and I think we're
all ready to go. So let's go back and
review our read module, so we do our
imports, we added our quit, we returned
results this part was the same Peter
buffer we loop, we read into our buffer,
if there's zero bytes we break otherwise
we return
number of bites if this an error we
break. We need to send the buffer that we
create out of the slice to the stats
thread in the future, or send an empty
buffer at the end, right now we'll unlock
our quit we'll change it to true to let
everyone else know we're done, and we'll
exit. Next up we have our stats thread.
This one won't take nearly as long. Start
with our imports we need the I/O result
because we don't have it yet we want to
return it. We want the same arc and mutex
imports as before, then we'll keep silent
but remove all of the other arguments,
and add our quit argument, again that's
an arc mutex of a boolean, and then we
want this to return the same thing as
the other loops. So it's a result of an
empty, which means we need to return an
ok empty at the end. And then we can go
back up to the top and change this to a
declaration of immutable total bytes, and
start it at 0, now let's make our loop so
put our loop here, close it down at the
end of the print, and then the first
thing that we need to do is receive our
bytes from the read thread. So let's put
a placeholder here to remind us of that
because we're not doing it in this video.
And since we're not receiving those
bytes let's make an empty buffer on the
same type that we'd expect to receive so
we can compile. Now we can actually count
the bytes in our buffer though it'll
just be 0 right now, add that to our
total bytes, if not silent is still
correct, this print is correct, but this
last one that goes outside the loop, and
we don't need a boolean anymore guarding
it because we know we're outside the
loop. But once we've counted our data we
do need to pass it on to the write loop,
so that it can write it out to standard
out or a file, and finally since we don't
have an actual buffer to look for a 0
length, let's look at quit and see if
it's time to quit out of this loop, will
break out of this loop due to either the
break or the end of the loop, quit will
go out of scope which will release the
lock. So let's recap we imported our can
mutex, we edit our quit parameter we
return an empty result, we
our total bytes, we start our loop, we
need to receive the vector of bytes,
we've got a placeholder for that, we
count the bytes, we increment total bytes,
if we're not silent we print out our
stats, we need to send the vector over
the write loop, and then we quit and
we've got our final print. And now we can
go to our write module. This is the last
module that we'll need to change, great
we're almost there. Start with our
imports≤ let's get our arc and mutex
like we did in our others, and then let's
go remove this buffer, and add our quit
parameter, it's the same as before arc
mutex bool. Then we'll go modify a result
to be an empty result, and then of course
our okay's need to become empty, so we
take out this false, take out this true,
let's fix these comments so they
represent reality, so this has stopped
the program cleanly. This is just our
success case so we don't need a comment.
This block of code for the writer all
looks correct, I don't think we need to
change anything here. So let's start
below it, let's create our loop and then
put the logic inside of it. The first
thing we need to do in our write loop is
receive the bytes that we need to write.
Of course we are not doing that in this
video, so we will put a placeholder for
that, and then we'll create an empty
buffer so that we can compile and run
our code. All right now, write doesn't
need to do anything if we haven't
received anything, so the first thing we
want to do with write is check if it's
time to quit. So for this video we'll
check our quit once again, but
this will last to the end of the block,
and that could lock through a long write
if we don't quit,
so we're gonna put quit inside of its
own scope, that way at the end of this
scope quit will go out of scope and
release the lock. So we can move on. This
write looks correct, the handling of
these conditions look correct, the
returns look correct, I think we're all
done with write, so let's go back and
clean up main. All of this logic has
already been implemented, so I can simply
remove it. There is one last thing read
loop, stats loop, write loop, we didn't
rename them so let's go to our functions
and actually put underscore loop at the
end of each of them, otherwise I really
don't think it will compile. So now if I
go back to main the IDE should have
colored our tan function color, good. So
if we go there and add that semicolon
then the compiler will know what to do.
Let's go back to the terminal, clear the
screen and run it again. Good now we can
try some things out now remember the
read loop reads the data in, and once
it's read it all in it sets the quit
boolean to true it which causes the
other threads to exit. So that's all
we'll see, so if we echo hello to Cargo
run, we see zero because we didn't pass
any data to stats and everything exited
successfully. This isn't very exciting
output I'll concede, but this is
demonstrating what we wanted to
demonstrate. We are spawning three child
threads, they communicate shut down via a
mutex, and then they are all joined to
the main thread which exits. In the next
video, we will use MPSC channels to
communicate between threads.