pijul/pijul - Change SN7AGY6SLIWVKLBHQQ2MZ7VYY4BBFKROC5F3M6FVVFIGT25QNX7AC

Making `pijul lock` robust to kill signals

Created by pmeunier on January 16, 2021

SN7AGY6SLIWVKLBHQQ2MZ7VYY4BBFKROC5F3M6FVVFIGT25QNX7AC

Dependencies

In channels

main

In discussions

#269 clone is broken

#301 Fix tests compilation failures

#333 Typo in help output

Change contents

Replacement in pijul/src/repository/unix_lock.rs at line 31 [4.1956]

B:BD[3.462] → [2.162:203]

        s.read_exact(&mut [0u8]).await?;

[3.462]

[3.491]

        if let Err(e) = s.read_exact(&mut [0u8]).await {
            process.wait().await?;
            return Err(e);
        }

Insertion in pijul/src/commands/lock.rs at line 1 [4.7180]
[4.7180]
[4.7245]
```
use clap::Clap;
use fs2::FileExt;
use std::fs::OpenOptions;
```
Replacement in pijul/src/commands/lock.rs at line 8 [4.7180]
B:BD[4.7365] → [4.7365:7415]
```
use tokio::sync::{watch, Mutex};
use clap::Clap;
```
[4.7365]
[4.7415]
```
use tokio::sync::{mpsc, watch, Mutex};
```

Insertion in pijul/src/commands/lock.rs at line 23 [4.7180]

[3.877]

[4.7905]

        }
        self.run_socket().await
    }
    async fn run_socket(self) -> Result<(), anyhow::Error> {
        let lock = self.path.with_extension("lock");
        let file = OpenOptions::new().write(true).create(true).open(&lock)?;
        if file.try_lock_exclusive().is_err() {
            return Ok(());

Insertion in pijul/src/commands/lock.rs at line 33 [4.7180]
[4.7915]
[3.878]
```
        std::fs::remove_file(&self.path).unwrap_or(());
```

Deletion in pijul/src/commands/lock.rs at line 41 [4.7180]

∅:D[3.1066] → [4.7999:8149]

B:BD[4.7999] → [4.7999:8149]

        let n_clients = Arc::new(Mutex::new(0usize));
        let file = Arc::new(self.path.clone());
        let muttxn = Arc::new(Mutex::new(()));

Replacement in pijul/src/commands/lock.rs at line 42 [4.7180]

B:BD[4.8210] → [4.8210:8358]

        let tx = std::sync::Arc::new(tx);
        let clock = Arc::new(Mutex::new(0usize));
        let txn_counter = Arc::new(Mutex::new(0usize));

[4.8210]

[4.8419]

        let (done_tx, mut done_rx) = mpsc::channel(1);
        let file = Arc::new(self.path);
        let locks = Locks {
            n_clients: Arc::new(Mutex::new(0usize)),
            muttxn: Arc::new(Mutex::new(())),
            tx: Arc::new(tx),
            active_at_last_commit,
            file: file.clone(),
            clock: Arc::new(Mutex::new(0usize)),
            txn_counter: Arc::new(Mutex::new(0usize)),
            done_tx,
        };

Replacement in pijul/src/commands/lock.rs at line 55 [4.7180]

B:BD[4.8434] → [4.8434:8479]

B:BD[4.8768] → [4.8768:8885]

B:BD[4.9272] → [4.9272:9496]

B:BD[4.9565] → [4.9565:9720]

B:BD[4.9765] → [4.9765:11259]

B:BD[4.11340] → [4.11340:12167]

B:BD[4.12167] → [3.1067:1137]

∅:D[3.1137] → [4.12248:12456]

B:BD[4.12248] → [4.12248:12456]

B:BD[4.12879] → [4.12879:13197]

∅:D[5.349] → [4.13531:13718]

B:BD[4.13531] → [4.13531:13718]

            let x = listener.accept().await;
            match x {
                Ok((mut stream, _addr)) => {
                    *n_clients.lock().await += 1;
                    let file = file.clone();
                    let muttxn = muttxn.clone();
                    let mut active_at_last_commit = active_at_last_commit.clone();
                    let clock = clock.clone();
                    let txn_counter = txn_counter.clone();
                    let n_clients = n_clients.clone();
                    let tx = tx.clone();
                    tokio::spawn(async move {
                        let mut t = [0u8];
                        while let Ok(n) = stream.read(&mut t).await {
                            if n == 0 {
                                break;
                            }
                            if t[0] == 1 {
                                // muttxn
                                let lock = if let Ok(guard) = muttxn.try_lock() {
                                    guard
                                } else {
                                    stream.write_all(&[LOCKED]).await.unwrap_or(());
                                    muttxn.lock().await
                                };
                                while *active_at_last_commit.borrow() > 0 {
                                    stream.write_all(&[LOCKED]).await.unwrap_or(());
                                    active_at_last_commit.changed().await.unwrap();
                                }
                                stream.write_all(&[ACK]).await.unwrap_or(());
                                if let Ok(n) = stream.read(&mut t).await {
                                    if n == 0 {
                                        break;
                                    }
                                    if t[0] == COMMIT {
                                        // commit
                                        let mut clock = clock.lock().await;
                                        *clock += 1;
                                        let counter = *txn_counter.lock().await;
                                        tx.send(counter).unwrap();
                                    }
                                }
                                std::mem::drop(lock);
                            } else {
                                // txn
                                *txn_counter.lock().await += 1;
                                let start_date = *clock.lock().await;
                                stream.write_all(&[ACK]).await.unwrap_or(());
                                let n = stream.read(&mut t).await.unwrap_or(0);
                                if n == 0 {
                                    break;
                                }
                                *txn_counter.lock().await -= 1;
                                if start_date < *clock.lock().await {
                                    let last = *active_at_last_commit.borrow();
                                    tx.send(last - 1).unwrap();
                                }
                            }
                        }
                        if *n_clients.lock().await == 1 {
                            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                            if *n_clients.lock().await == 1 {
                                std::fs::remove_file(file.as_ref()).unwrap_or(());
                                std::process::exit(0)
                            }
                        }
                        *n_clients.lock().await -= 1
                    });

[4.8434]

[4.13718]

            tokio::select! {
                x = tokio::time::timeout(std::time::Duration::from_secs(1), listener.accept()) => {
                    if let Ok(Ok((stream, _))) = x {
                        accept(stream, locks.clone()).await
                    } else {
                        break
                    }
                },
                _ = done_rx.recv() => {
                    break

Deletion in pijul/src/commands/lock.rs at line 66 [4.7180]
B:BD[4.13736] → [4.13736:13769]
```
                Err(_) => break,
```
Replacement in pijul/src/commands/lock.rs at line 69 [4.7180]
B:BD[4.14042] → [4.14042:14072]
```
        std::process::exit(0)
```
[4.13852]
[4.14072]
```
        Ok(())
```

Insertion in pijul/src/commands/lock.rs at line 71 [4.7180]

[4.14078]

}
#[derive(Clone)]
struct Locks {
    n_clients: Arc<Mutex<usize>>,
    file: Arc<PathBuf>,
    muttxn: Arc<Mutex<()>>,
    clock: Arc<Mutex<usize>>,
    txn_counter: Arc<Mutex<usize>>,
    active_at_last_commit: watch::Receiver<usize>,
    tx: Arc<watch::Sender<usize>>,
    done_tx: mpsc::Sender<()>,

Insertion in pijul/src/commands/lock.rs at line 84 [4.7180]

[4.14080]


async fn accept(mut stream: tokio::net::UnixStream, mut locks: Locks) {
    *locks.n_clients.lock().await += 1;
    tokio::spawn(async move {
        let mut t = [0u8];
        while let Ok(n) = stream.read(&mut t).await {
            if n == 0 {
                break;
            }
            if t[0] == 1 {
                // muttxn
                let lock = if let Ok(guard) = locks.muttxn.try_lock() {
                    guard
                } else {
                    stream.write_all(&[LOCKED]).await.unwrap_or(());
                    locks.muttxn.lock().await
                };
                while *locks.active_at_last_commit.borrow() > 0 {
                    stream.write_all(&[LOCKED]).await.unwrap_or(());
                    locks.active_at_last_commit.changed().await.unwrap();
                }
                stream.write_all(&[ACK]).await.unwrap_or(());
                if let Ok(n) = stream.read(&mut t).await {
                    if n == 0 {
                        break;
                    }
                    if t[0] == COMMIT {
                        // commit
                        let mut clock = locks.clock.lock().await;
                        *clock += 1;
                        let counter = *locks.txn_counter.lock().await;
                        locks.tx.send(counter).unwrap();
                    }
                }
                std::mem::drop(lock)
            } else {
                // txn
                *locks.txn_counter.lock().await += 1;
                let start_date = *locks.clock.lock().await;
                stream.write_all(&[ACK]).await.unwrap_or(());
                let n = stream.read(&mut t).await.unwrap_or(0);
                if n == 0 {
                    break;
                }
                *locks.txn_counter.lock().await -= 1;
                if start_date < *locks.clock.lock().await {
                    let last = *locks.active_at_last_commit.borrow();
                    locks.tx.send(last - 1).unwrap();
                }
            }
        }
        if *locks.n_clients.lock().await == 1 {
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            if *locks.n_clients.lock().await == 1 {
                locks.done_tx.send(()).await.unwrap();
                return;
            }
        }
        *locks.n_clients.lock().await -= 1
    });
}

Replacement in pijul/Cargo.toml at line 4 [6.196374]
B:BD[6.196462] → [4.16737:16764]
```
version = "1.0.0-alpha.33"
```
[6.196462]
[6.196488]
```
version = "1.0.0-alpha.35"
```
Insertion in pijul/Cargo.toml at line 87 [6.196374]
[7.2300]
[8.309]
```
fs2 = "0.4"
```

Making `pijul lock` robust to kill signals

Dependencies

In channels

In discussions

Change contents

Replacement in pijul/src/repository/unix_lock.rs at line 31 [4.1956]

Insertion in pijul/src/commands/lock.rs at line 1 [4.7180]

Replacement in pijul/src/commands/lock.rs at line 8 [4.7180]

Insertion in pijul/src/commands/lock.rs at line 23 [4.7180]

Insertion in pijul/src/commands/lock.rs at line 33 [4.7180]

Deletion in pijul/src/commands/lock.rs at line 41 [4.7180]

Replacement in pijul/src/commands/lock.rs at line 42 [4.7180]

Replacement in pijul/src/commands/lock.rs at line 55 [4.7180]

Deletion in pijul/src/commands/lock.rs at line 66 [4.7180]

Replacement in pijul/src/commands/lock.rs at line 69 [4.7180]

Insertion in pijul/src/commands/lock.rs at line 71 [4.7180]

Insertion in pijul/src/commands/lock.rs at line 84 [4.7180]

Replacement in pijul/Cargo.toml at line 4 [6.196374]

Insertion in pijul/Cargo.toml at line 87 [6.196374]