ilyagr/pijul - Change G65S7FAW4OHBGC6ROVPPZ6PPACKOFEM4KSMAWMYKNQF5OLVDJR2QC

Version bump and cleanup

Created by pmeunier on February 25, 2021

G65S7FAW4OHBGC6ROVPPZ6PPACKOFEM4KSMAWMYKNQF5OLVDJR2QC

Dependencies

In channels

main

Change contents

Replacement in pijul-macros/Cargo.toml at line 4 [3.21011]
B:BD[3.21091] → [4.7647:7665]
```
version = "0.3.0"
```
[3.21091]
[3.21109]
```
version = "0.4.0"
```
File deletion: repository
BF:BFD[3.21414] → [5.1927:1939]
BF:BFD[5.1939] → [5.1940:1940]

File deletion: basic_lock.rs

BF:BFD[5.1940] → [5.3644:3659]

BF:BFD[5.3659] → [5.3660:3660]

B:BD[5.3660] → [5.3661:4572]

use fs2::FileExt;
use std::fs::OpenOptions;
use std::path::Path;
pub struct TxnLock {
    file: std::fs::File,
}
pub struct MutTxnLock {
    file: std::fs::File,
}
pub async fn mut_txn<P: AsRef<Path>>(file: P) -> Result<MutTxnLock, anyhow::Error> {
    let file = OpenOptions::new().write(true).create(true).open(file)?;
    file.lock_exclusive()?;
    Ok(MutTxnLock { file })
}
pub async fn txn<P: AsRef<Path>>(file: P) -> Result<TxnLock, anyhow::Error> {
    let file = OpenOptions::new().write(true).create(true).open(file)?;
    file.lock_shared()?;
    Ok(TxnLock { file })
}
impl MutTxnLock {
    pub async fn commit(&mut self) -> Result<(), anyhow::Error> {
        Ok(self.file.unlock()?)
    }
}
impl Drop for MutTxnLock {
    fn drop(&mut self) {
        self.file.unlock().unwrap_or(())
    }
}
impl Drop for TxnLock {
    fn drop(&mut self) {
        self.file.unlock().unwrap_or(())
    }
}

File deletion: unix_lock.rs

BF:BFD[5.1940] → [5.1941:1955]

BF:BFD[5.1955] → [5.1956:1956]

B:BD[5.1956] → [5.1957:1975]

B:BD[5.1994] → [5.1994:2286]

B:BD[5.2286] → [6.0:227]

B:BD[6.227] → [7.0:61]

∅:D[7.61] → [6.286:389]

B:BD[6.286] → [6.286:389]

∅:D[7.161] → [6.389:462]

B:BD[6.389] → [6.389:462]

B:BD[6.462] → [8.0:129]

∅:D[8.129] → [6.491:539]

∅:D[7.203] → [6.491:539]

B:BD[6.491] → [6.491:539]

∅:D[6.539] → [5.2597:2598]

B:BD[5.2597] → [5.2597:2598]

B:BD[5.2598] → [6.540:692]

∅:D[6.692] → [5.2724:3050]

B:BD[5.2724] → [5.2724:3050]

B:BD[5.3050] → [6.693:790]

∅:D[6.790] → [5.3267:3643]

B:BD[5.3267] → [5.3267:3643]

B:BD[6.389] → [7.62:161]

use anyhow::bail;
use std::path::Path;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::UnixStream;
pub struct TxnLock {
    _stream: UnixStream,
}
pub struct MutTxnLock {
    stream: UnixStream,
}
const TXN_BEGIN: u8 = 0;
const MUT_TXN_BEGIN: u8 = 1;
const COMMIT: u8 = 2;
const ACK: u8 = 3;
async fn start_stream_if_needed(file: &Path) -> Result<UnixStream, std::io::Error> {
    if let Ok(stream) = UnixStream::connect(file).await {
        Ok(stream)
    } else {
        let cmd = std::env::args().next().unwrap();
        let mut process = tokio::process::Command::new(&cmd)
            .args(&["lock", file.to_str().unwrap()])
            .stdout(std::process::Stdio::piped())
            .spawn()?;
        let s = process.stdout.as_mut().unwrap();
        if let Err(e) = s.read_exact(&mut [0u8]).await {
            process.wait().await?;
            return Err(e);
        }
        UnixStream::connect(file).await
    }
}
pub async fn mut_txn<P: AsRef<Path>>(file: P) -> Result<MutTxnLock, anyhow::Error> {
    let mut stream = start_stream_if_needed(file.as_ref()).await?;
    stream.writable().await?;
    stream.write_all(&[MUT_TXN_BEGIN]).await?;
    let mut t = [0u8];
    stream.read_exact(&mut t).await?;
    if t[0] == ACK {
        Ok(MutTxnLock { stream })
    } else {
        bail!("Pristine locked")
    }
}
pub async fn txn<P: AsRef<Path>>(file: P) -> Result<TxnLock, anyhow::Error> {
    let mut stream = start_stream_if_needed(file.as_ref()).await?;
    stream.writable().await?;
    stream.write_all(&[TXN_BEGIN]).await?;
    let mut t = [0u8];
    stream.read_exact(&mut t).await?;
    if t[0] == ACK {
        Ok(TxnLock { _stream: stream })
    } else {
        bail!("Pristine locked")
    }
}
impl MutTxnLock {
    pub async fn commit(&mut self) -> Result<(), anyhow::Error> {
        self.stream.write_all(&[COMMIT]).await?;
        Ok(())
    }
}
            .stdin(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())

File deletion: lock.rs

BF:BFD[3.93386] → [5.7170:7179]

BF:BFD[5.7179] → [5.7180:7180]

B:BD[5.7180] → [8.130:190]

∅:D[8.190] → [5.7245:7365]

B:BD[5.7245] → [5.7245:7365]

B:BD[5.7365] → [8.191:230]

∅:D[8.230] → [5.7415:7613]

B:BD[5.7415] → [5.7415:7613]

B:BD[5.7613] → [6.791:877]

B:BD[6.877] → [8.231:546]

∅:D[8.546] → [5.7905:7915]

∅:D[6.877] → [5.7905:7915]

B:BD[5.7905] → [5.7905:7915]

∅:D[8.603] → [6.878:933]

B:BD[5.7915] → [6.878:933]

∅:D[9.137] → [5.7979:7999]

∅:D[6.933] → [5.7979:7999]

B:BD[5.7979] → [5.7979:7999]

B:BD[5.7999] → [6.934:1066]

B:BD[5.8149] → [5.8149:8210]

B:BD[5.8210] → [8.604:1059]

∅:D[8.1059] → [5.8419:8434]

B:BD[5.8419] → [5.8419:8434]

B:BD[5.8434] → [8.1060:1189]

B:BD[8.1189] → [10.0:411]

∅:D[10.411] → [8.1361:1468]

B:BD[8.1361] → [8.1361:1468]

∅:D[8.1468] → [5.13718:13736]

B:BD[5.13718] → [5.13718:13736]

B:BD[5.13769] → [5.13769:13852]

B:BD[5.13852] → [8.1469:1484]

∅:D[8.1484] → [5.14072:14078]

B:BD[5.14072] → [5.14072:14078]

∅:D[8.1789] → [5.14078:14080]

B:BD[5.14078] → [5.14078:14080]

B:BD[5.14080] → [8.1790:4166]

B:BD[5.14078] → [8.1485:1789]

B:BD[5.7915] → [8.547:603]

use clap::Clap;
use fs2::FileExt;
use std::fs::OpenOptions;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::UnixListener;
use tokio::sync::{mpsc, watch, Mutex};
#[derive(Clap, Debug)]
pub struct Lock {
    path: PathBuf,
}
const COMMIT: u8 = 2;
const ACK: u8 = 3;
const LOCKED: u8 = 4;
impl Lock {
    pub async fn run(self) -> Result<(), anyhow::Error> {
        if let Some(p) = self.path.parent() {
            std::fs::create_dir_all(p)?
        }
        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(());
        }
        let listener = UnixListener::bind(&self.path);
        println!();
        let listener = if let Ok(listener) = listener {
            listener
        } else {
            return Ok(());
        };
        let (tx, active_at_last_commit) = watch::channel(0);
        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,
        };
        loop {
            tokio::select! {
                x = tokio::time::timeout(std::time::Duration::from_secs(1), listener.accept()) => {
                    match x {
                        Err(_) => {
                            if *locks.n_clients.lock().await == 0 {
                                break
                            }
                        }
                        Ok(Ok((stream, _))) => {
                            accept(stream, locks.clone()).await
                        }
                        Ok(Err(_)) => break
                    }
                },
                _ = done_rx.recv() => {
                    break
                }
            }
        }
        std::fs::remove_file(file.as_ref()).unwrap_or(());
        Ok(())
    }
}
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
    });
}
}
#[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<()>,
        std::fs::remove_file(&self.path).unwrap_or(());

Replacement in pijul/Cargo.toml at line 4 [3.196374]
B:BD[3.196462] → [2.0:27]
```
version = "1.0.0-alpha.38"
```
[3.196462]
[3.196488]
```
version = "1.0.0-alpha.39"
```
Deletion in pijul/Cargo.toml at line 32 [3.196374]
B:BD[3.197171] → [5.16765:16789]
```
"src/commands/lock.rs",
```
Deletion in pijul/Cargo.toml at line 35 [3.196374]
B:BD[3.197209] → [5.16790:16853]
```
"src/repository/unix_lock.rs",
"src/repository/basic_lock.rs",
```

Replacement in pijul/Cargo.toml at line 51 [3.196374]

B:BD[3.197438] → [2.28:118]

libpijul = { path = "../libpijul", version = "1.0.0-alpha.34", features = [ "tarball" ] }

[3.197438]

[3.197523]

libpijul = { path = "../libpijul", version = "1.0.0-alpha.35", features = [ "tarball" ] }

Replacement in libpijul/Cargo.toml at line 4 [3.1020742]
B:BD[3.1020894] → [2.146:173]
```
version = "1.0.0-alpha.34"
```
[3.1020894]
[11.27]
```
version = "1.0.0-alpha.35"
```

Replacement in libpijul/Cargo.toml at line 107 [3.1020742]

B:BD[3.1023195] → [12.2127:2190]

pijul-macros = { path = "../pijul-macros", version = "0.3.0" }

[3.1023195]

[3.1023218]

pijul-macros = { path = "../pijul-macros", version = "0.4.0" }

Replacement in Cargo.lock at line 948 [3.1030385]
B:BD[13.560] → [2.174:201]
```
version = "1.0.0-alpha.34"
```
[13.560]
[13.729]
```
version = "1.0.0-alpha.35"
```
Replacement in Cargo.lock at line 1310 [3.1030385]
B:BD[3.1064071] → [2.202:229]
```
version = "1.0.0-alpha.38"
```
[3.1064071]
[3.1064097]
```
version = "1.0.0-alpha.39"
```
Replacement in Cargo.lock at line 1355 [3.1030385]
B:BD[3.1064574] → [4.142288:142306]
```
version = "0.3.0"
```
[3.1064574]
[3.1064592]
```
version = "0.4.0"
```

Version bump and cleanup

Dependencies

In channels

Change contents

Replacement in pijul-macros/Cargo.toml at line 4 [3.21011]

File deletion: repository

File deletion: basic_lock.rs

File deletion: unix_lock.rs

File deletion: lock.rs

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

Deletion in pijul/Cargo.toml at line 32 [3.196374]

Deletion in pijul/Cargo.toml at line 35 [3.196374]

Replacement in pijul/Cargo.toml at line 51 [3.196374]

Replacement in libpijul/Cargo.toml at line 4 [3.1020742]

Replacement in libpijul/Cargo.toml at line 107 [3.1020742]

Replacement in Cargo.lock at line 948 [3.1030385]

Replacement in Cargo.lock at line 1310 [3.1030385]

Replacement in Cargo.lock at line 1355 [3.1030385]