refactor: Shared to use internal mutability

[akaladarshi]

Fixes: #170

This PR changes:

• Now the Shared struct internally handles the mutex.
• Shared struct now has inner: Arc<Mutex<SharedInner>>

[elenaf9]

Thank you for your PR @akaladarshi!

Some comments.

[elenaf9]

parking_lot::Mutex doesn't support re-entrant locking, thus a nested call to this would cause a deadlock.
To avoid this, the function should take &mut self.

[akaladarshi]

Thanks for pointing this out!

I need a suggestion regarding the design choices in the code snippet below. We are frequently using Pin<&mut Self> in connection.rs. So if we change fn with _mut to take &mut self then, we would run into borrow checkers issue.

So to fix it, is the below snippet a good choice or we could do something like take a mutexGuard on sharedInner or is there any better way?.

Here’s the code for reference (stream.rs -> fn poll_next):

        let conn_id = self.conn.clone();
        let stream_id = self.id.clone();
        let shared = &mut self.shared;
        let poll_state = shared.with_mut(|inner| {
            if let Some(bytes) = inner.buffer.pop() {
                let off = bytes.offset();
                let mut vec = bytes.into_vec();
                if off != 0 {
                    // This should generally not happen when the stream is used only as
                    // a `futures::stream::Stream` since the whole point of this impl is
                    // to consume chunks atomically. It may perhaps happen when mixing
                    // this impl and the `AsyncRead` one.
                    log::debug!(
                        "{}/{}: chunk has been partially consumed",
                        conn_id,
                        stream_id,
                    );
                    vec = vec.split_off(off)
                }
                return Poll::Ready(Some(Ok(Packet(vec))));
            }

            // Buffer is empty, let's check if we can expect to read more data.
            if !inner.state.can_read() {
                log::debug!("{}/{}: eof", conn_id, stream_id);
                return Poll::Ready(None); // stream has been reset
            }

            // Since we have no more data at this point, we want to be woken up
            // by the connection when more becomes available for us.
            inner.reader = Some(cx.waker().clone());
            Poll::Pending
        });

[akaladarshi]

Updated the with_mut to use &mut.

I added another lock() function, which will allows us to use simple lock wherever it's necessary and also keeping the with_mut so we can use whichever we need.

[elenaf9]

I need a suggestion regarding the design choices in the code snippet below. We are frequently using Pin<&mut Self> in connection.rs. So if we change fn with_mut to take &mut self then, we would run into borrow checkers issue.

You could use Pin::get_mut and then destructure Self to access the distinct fields, which would satisfy the borrow checker:

        let Self {
            conn, shared, id, ..
        } = self.get_mut();
        shared.with_mut(|inner| {
            inner.update_state(*conn, *id, State::SendClosed);
        });

I added another lock() function, which will allows us to use simple lock wherever it's necessary and also keeping the with_mut so we can use whichever we need.

lock also needs to take &mut self for the same reasons.
Generally, I am not a fan of having both lock and with_mut since they are just two ways of doing the same thing. I am okay with either of them, but we should decide for one.