In this video, we will use MPSC channels
to communicate between threads. Here are
the four threads of execution in our
program. We start with the main thread
that launches the three child threads,
the reed thread, the stats thread, and the
write thread. In this video, we're gonna
set up two channels a channel to
communicate from the read thread to the
stats thread, and a channel to
communicate from the stats thread to the
write thread. We are going to send
buffers of u8 across channels, and use an
empty buffer as a sentinel to stop,
instead of using an atomically reference
counted mutex guarded boolean to
indicate though we need to stop. Let's
start by importing the MPSC module
instead of arc and mutex, then let's
go down and create our channels. We will
receive a tuple containing a sender and
a receiver side of the channel. I'm going
to name the channels after the thread
that is receiving the data, and then I'll
use TX to mean this is the transmitting
to that thread, and rx to mean the
receiving inside of that thread. So stats
TX will be used to transmit something to
the stats thread, and stats rx will be
used to receive something in the stats
thread. We also receive data in the write
thread so let's make that channel. We
have our transmit to the write thread
and receive in the write thread ends of
the channel. I should mention that MPSC
stands for a multiple producer single
consumer, so you can actually take that
sending side the TX sides of the channel
and clone them, and send them to
different places, different threads, and
use them all to write to the same
channel, but you can only have one
receiving side of the channel.
So now let's clean up our mutex and
replace them in our arguments with our
channels, updating our function
signatures we'll fix these errors, but
first let's talk about how this data
flows through. We give the read thread a
sender to the stats thread, we give the
stats thread its own
receiver and the sender to the write
thread, and then we give the write thread
its own receiver. So now let's go to the
reed thread and update it to accept that
sender to the stats thread. So first
thing we need to do is go to our imports
and update it, we want the sender from
the MPSC module, then we can go down and
replace our quit argument with our stats
sender, which is a Vec of u8. And now we
can go down and actually do our two dues.
First we need to send this empty buffer
so stats tx send will send that across the
channel. Now sending could fail so this
returns a result which you need to do
something with. Let's look and see if
it's an error, and if it is an error
let's just exit out cleanly, because that
means that our stats thread has been
shut down, so we're probably shutting
down. And then of course, if we've run out
of data to read we know we need to shut
down. So let's send that empty buffer
over to the stats thread, but in this
case we'll ignore the result all
together because we're shutting down. Ok,
let's go back up and recap what we did.
We imported sender, we updated our
argument, we came down here, we send our
buffer over to the stats thread, we check
to see if it's an error we exit out
cleanly, and then we'll send our empty
vector if we're done. Now we can go on to
the stats that I didn't handle that.
First our imports this time we'll need
both the receiver and the sender, those
come from the MPSC module, and then we
can go down and add both of those to our
arguments. Our stats receiver is a
receiver of Vec u8, and then of course
our write sender to send to the write
thread as a sender of Vec u8, whoa this
line is getting really long, let's go see
what Cargo fmt would do to this line.
I'll just switch over to the terminal
and run it real quick, and then switch
back and now you can see the vertical
formatting style
for argument lists. Notice it added
an extra comment after the write sender,
now let's go down and implement our
to-do. Let's actually receive the buffer,
we are going to call receive on our
channel receiver, which is a blocking
call that returns a result once we get
data, which we're just going to end wrap
in this case, and then we're going to
grab the number of bytes in the buffer,
I'll show you why in just a second.
I'll change total bytes to increment by
the number of bytes that I've just
collected, and then come down and
implement the other to do, which is
sending the vector over to the write
thread. So in this case, I know it's
returning the result, I know I want to
check the error, so I'm just going to
start with my yes statement and I will
break out cleanly if we are not able to
send. We know the pipes are gonna break
as we're trying to write to them. What
you may not have realized is that send
consumes the value that you give it, so
we no longer have buffer to look at, and
write here I want to see how many bytes
I received and break out, but I can't
look at buffer dot Len, that's why I
saved num bytes, so I can look at it now.
So let's check to see if num bytes is 0
and if it is will break out. Finally,
let's go fix a bug that I notice that we
ended up with in the last section. We're
only supposed to print if we're not
silent, so I'll just surround this with
an if not silent, now we should be good
to go on our stats module. Quick recap, we
updated our imports we updated our
argument list to get our receiver and
our sender,
we receive a buffer from the read thread,
we count the number of bytes in it, and
then we send it over to the write thread,
break if there's an error, and then if we
didn't actually receive anything we also
need to break out, and we fix the bug
with our output. And that's good for the
stats module. Let's move over to the
write module again. Once again we're
going to begin by removing arc and mutex
in favor of our MPSC receiver, but we got
to remember put MPSC on there or it
won't come in. Alright, let's go and
change our argument this is the write
receiver, which is a receiver of a Vec
of u8, and then we can hop write down to
our two
do, and receive our buffer just like we
did before. Once again we'll use the
blocking receive and just unwrap on
error. Instead of this quit block here
let's look for empty buffer Sentinel,
so if buffer is empty then we're going
to break out of our loop and shut down
cleanly. And I believe that is all we
have left to do. Let's go give it a try.
Let's start by running Cargo check again,
great no errors,
how about Cargo Clippy any style changes,
nothing great. So let's run Cargo fmt
just in case, and then let's clear the
screen and do some testing. First let's
do our tried and true echo hello to our
program, and see that it outputs our six
in our hello, there it goes great. Let's
try to isolate the two outputs to make
sure that we don't have that extra
newline bug, remember our - - okay. So
there's our six if we're out our output,
what if we do silent okay, there's our
hello no six looks correct I don't see
any stray new lines this time. Let's pipe
yes to our program and just redirect our
output to dev null for a while, and see
if we can just accept a bunch of input,
great looks like it's running just fine.
So let's change that up a little bit,
instead let's pipe over to head and try
breaking our pipe and making sure that
we exit cleanly.
Okay there we go, everything appears to
be working as intended. In the next video,
we will refactor our code to use
crossbeam channels instead of MPSC
channels.