diff --git a/ipcd/.gitignore b/ipcd/.gitignore new file mode 100644 index 0000000000..899ce2fc38 --- /dev/null +++ b/ipcd/.gitignore @@ -0,0 +1,3 @@ +/target +**/*.rs.bk +/Xargo.toml diff --git a/ipcd/.gitlab-ci.yml b/ipcd/.gitlab-ci.yml new file mode 100644 index 0000000000..d019614988 --- /dev/null +++ b/ipcd/.gitlab-ci.yml @@ -0,0 +1,26 @@ +image: "redoxos/redoxer" + +stages: + - build + - test + +cache: + paths: + - target/ + +build:redox: + stage: build + script: redoxer build --verbose + +test:redox: + stage: test + dependencies: + - build:redox + script: + - redoxer build --verbose --examples + - redoxer exec --folder . -- sh -- ./redoxer.sh + event + main + unnamed + shm_demo + # FIXME: shm is currently broken diff --git a/ipcd/Cargo.lock b/ipcd/Cargo.lock new file mode 100644 index 0000000000..bcf353548b --- /dev/null +++ b/ipcd/Cargo.lock @@ -0,0 +1,76 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 3 + +[[package]] +name = "bitflags" +version = "2.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b048fb63fd8b5923fc5aa7b340d8e156aec7ec02f0c78fa8a6ddc2613f6f71de" + +[[package]] +name = "ipcd" +version = "0.1.0" +dependencies = [ + "libredox", + "redox-daemon", + "redox-scheme", + "redox_event", + "redox_syscall", +] + +[[package]] +name = "libc" +version = "0.2.160" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0b21006cd1874ae9e650973c565615676dc4a274c965bb0a73796dac838ce4f" + +[[package]] +name = "libredox" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0ff37bd590ca25063e35af745c343cb7a0271906fb7b37e4813e8f79f00268d" +dependencies = [ + "bitflags", + "libc", + "redox_syscall", +] + +[[package]] +name = "redox-daemon" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7fbdeffdb03cf2961b211a2023e683d71f2c225ea93404da5dc34b0dc94f0341" +dependencies = [ + "libc", + "libredox", +] + +[[package]] +name = "redox-scheme" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "95624e20d2c1808f7ca0820720d30aad9f5d2fc404e1ef379431ad7a790c3f7e" +dependencies = [ + "libredox", + "redox_syscall", +] + +[[package]] +name = "redox_event" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69609faa5d5992247a4ef379917bb3e39be281405d6a0ccd4f942429400b956f" +dependencies = [ + "bitflags", + "libredox", +] + +[[package]] +name = "redox_syscall" +version = "0.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b6dfecf2c74bce2466cabf93f6664d6998a69eb21e39f4207930065b27b771f" +dependencies = [ + "bitflags", +] diff --git a/ipcd/Cargo.toml b/ipcd/Cargo.toml new file mode 100644 index 0000000000..89199359fa --- /dev/null +++ b/ipcd/Cargo.toml @@ -0,0 +1,11 @@ +[package] +edition = "2018" +name = "ipcd" +version = "0.1.0" + +[dependencies] +libredox = "0.1.3" +redox-daemon = "0.1.2" +redox_syscall = "0.5" +redox-scheme = "0.2.1" +redox_event = "0.4.0" diff --git a/ipcd/LICENSE b/ipcd/LICENSE new file mode 100644 index 0000000000..897e4bc3ee --- /dev/null +++ b/ipcd/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2018 Redox OS + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/ipcd/README.md b/ipcd/README.md new file mode 100644 index 0000000000..8f726664b9 --- /dev/null +++ b/ipcd/README.md @@ -0,0 +1,28 @@ +# ipcd + +A userspace daemon for interprocess communication. + +## Usage + +See the `examples` folder. + +Simply open `chan:` with O_CREAT where `` is any name you'd like to create a listener. +This listener can accept clients by calling `dup("listen")`. + +Open `chan:` without O_CREAT to connect. Now you can read and write between both streams. + +## How To Contribute + +To learn how to contribute to this system component you need to read the following document: + +- [CONTRIBUTING.md](https://gitlab.redox-os.org/redox-os/redox/-/blob/master/CONTRIBUTING.md) + +## Development + +To learn how to do development with this system component inside the Redox build system you need to read the [Build System](https://doc.redox-os.org/book/build-system-reference.html) and [Coding and Building](https://doc.redox-os.org/book/coding-and-building.html) pages. + +### How To Build + +To build this system component you need to download the Redox build system, you can learn how to do it on the [Building Redox](https://doc.redox-os.org/book/podman-build.html) page. + +This is necessary because they only work with cross-compilation to a Redox virtual machine, but you can do some testing from Linux. diff --git a/ipcd/examples/event.rs b/ipcd/examples/event.rs new file mode 100644 index 0000000000..6a42b45a78 --- /dev/null +++ b/ipcd/examples/event.rs @@ -0,0 +1,137 @@ +use std::{ + fs::File, + io::{self, prelude::*}, + os::unix::io::{AsRawFd, FromRawFd, RawFd} +}; + +fn from_syscall_error(error: syscall::Error) -> io::Error { + io::Error::from_raw_os_error(error.errno as i32) +} +fn nonblock(file: &File) -> io::Result<()> { + syscall::fcntl(file.as_raw_fd() as usize, syscall::F_SETFL, syscall::O_NONBLOCK) + .map(|_| ()) + .map_err(from_syscall_error) +} +fn dup(file: &File, buf: &str) -> io::Result { + let stream = syscall::dup(file.as_raw_fd() as usize, buf.as_bytes()).map_err(from_syscall_error)?; + Ok(unsafe { File::from_raw_fd(stream as RawFd) }) +} + +fn main() -> io::Result<()> { + let server = File::create("chan:hello_world")?; + + nonblock(&server)?; + + let mut event_file = File::open("event:")?; + let mut time_file = File::open(format!("time:{}", syscall::CLOCK_MONOTONIC))?; + + let mut time = syscall::TimeSpec::default(); + time_file.read(&mut time)?; + time.tv_sec += 1; + time_file.write(&time)?; + time.tv_sec += 2; + time_file.write(&time)?; + time.tv_sec += 2; + time_file.write(&time)?; + + const TOKEN_TIMER: usize = 0; + const TOKEN_STREAM: usize = 1; + const TOKEN_SERVER: usize = 2; + const TOKEN_CLIENT: usize = 3; + + event_file.write(&syscall::Event { + id: time_file.as_raw_fd() as usize, + flags: syscall::EVENT_READ, + data: TOKEN_TIMER + })?; + event_file.write(&syscall::Event { + id: server.as_raw_fd() as usize, + flags: syscall::EVENT_WRITE | syscall::EVENT_READ, + data: TOKEN_SERVER + })?; + + let mut event = syscall::Event::default(); + + println!("Testing accept events..."); + + event_file.read(&mut event)?; + assert_eq!(event.data, TOKEN_TIMER); + assert_eq!(event.flags, syscall::EVENT_READ); + println!("-> Timed out"); + + let mut client = File::open("chan:hello_world")?; + event_file.write(&syscall::Event { + id: client.as_raw_fd() as usize, + flags: syscall::EVENT_WRITE | syscall::EVENT_READ, + data: TOKEN_CLIENT + })?; + + event_file.read(&mut event)?; + assert_eq!(event.data, TOKEN_SERVER); + assert_eq!(event.flags, syscall::EVENT_WRITE | syscall::EVENT_READ); + println!("-> Accept event"); + + println!("Testing write events..."); + + let mut stream = dup(&server, "listen")?; + + event_file.read(&mut event)?; + assert_eq!(event.data, TOKEN_CLIENT); + assert_eq!(event.flags, syscall::EVENT_WRITE); + println!("-> Writable event"); + + event_file.write(&syscall::Event { + id: stream.as_raw_fd() as usize, + flags: syscall::EVENT_READ | syscall::EVENT_WRITE, + data: TOKEN_STREAM + })?; + + event_file.read(&mut event)?; + assert_eq!(event.data, TOKEN_STREAM); + assert_eq!(event.flags, syscall::EVENT_WRITE); + println!("-> Writable event"); + + event_file.read(&mut event)?; + assert_eq!(event.data, TOKEN_TIMER); + assert_eq!(event.flags, syscall::EVENT_READ); + println!("-> Timed out"); + + println!("Testing read events..."); + + client.write(b"a")?; + + let mut buf = [0; 5]; + + event_file.read(&mut event)?; + assert_eq!(event.data, TOKEN_STREAM); + assert_eq!(event.flags, syscall::EVENT_READ); + println!("-> Readable event"); + + assert_eq!(stream.read(&mut buf)?, 1); + assert_eq!(buf[0], b'a'); + + stream.write(b"b")?; + + event_file.read(&mut event)?; + assert_eq!(event.data, TOKEN_CLIENT); + assert_eq!(event.flags, syscall::EVENT_READ); + println!("-> Readable event"); + + assert_eq!(client.read(&mut buf)?, 1); + assert_eq!(buf[0], b'b'); + + drop(client); + + event_file.read(&mut event)?; + assert_eq!(event.data, TOKEN_STREAM); + println!("-> Readable event (EOF)"); + + assert_eq!(stream.read(&mut buf)?, 0); + + event_file.read(&mut event)?; + assert_eq!(event.data, TOKEN_TIMER); + println!("-> Timed out"); + + println!("Everything tested!"); + Ok(()) +} diff --git a/ipcd/examples/helloclient.rs b/ipcd/examples/helloclient.rs new file mode 100644 index 0000000000..95af217d0e --- /dev/null +++ b/ipcd/examples/helloclient.rs @@ -0,0 +1,11 @@ +use std::{ + fs::File, + io +}; + +fn main() -> io::Result<()> { + let mut client = File::open("chan:hello")?; + io::copy(&mut client, &mut io::stdout())?; + + Ok(()) +} diff --git a/ipcd/examples/helloserver.rs b/ipcd/examples/helloserver.rs new file mode 100644 index 0000000000..5a2df14095 --- /dev/null +++ b/ipcd/examples/helloserver.rs @@ -0,0 +1,20 @@ +use std::{ + fs::File, + io::{self, prelude::*}, + os::unix::io::{AsRawFd, FromRawFd, RawFd} +}; + +fn from_syscall_error(error: syscall::Error) -> io::Error { + io::Error::from_raw_os_error(error.errno as i32) +} + +fn main() -> io::Result<()> { + let server = File::create("chan:hello")?; + + loop { + let stream = syscall::dup(server.as_raw_fd() as usize, b"listen").map_err(from_syscall_error)?; + let mut stream = unsafe { File::from_raw_fd(stream as RawFd) }; + + stream.write(b"Hello World!\n")?; + } +} diff --git a/ipcd/examples/logserver.rs b/ipcd/examples/logserver.rs new file mode 100644 index 0000000000..a060ea30a4 --- /dev/null +++ b/ipcd/examples/logserver.rs @@ -0,0 +1,90 @@ +use std::{ + collections::HashMap, + fs::File, + io::{self, prelude::*}, + os::unix::io::{AsRawFd, FromRawFd, RawFd}, + str +}; + +fn from_syscall_error(error: syscall::Error) -> io::Error { + io::Error::from_raw_os_error(error.errno as i32) +} + +fn main() -> io::Result<()> { + let server = File::create("chan:log")?; + let mut event_file = File::open("event:")?; + + syscall::fcntl(server.as_raw_fd() as usize, syscall::F_SETFL, syscall::O_NONBLOCK) + .map(|_| ()) + .map_err(from_syscall_error)?; + event_file.write(&syscall::Event { + id: server.as_raw_fd() as usize, + data: 0, + flags: syscall::EVENT_READ | syscall::EVENT_WRITE, + })?; + + let mut clients = HashMap::new(); + let mut next_id = 1; + + 'outer: loop { + let mut event = syscall::Event::default(); + event_file.read(&mut event)?; + + if event.data == 0 { + println!("Listener recevied flags: {:?}", event.flags); + if event.flags & syscall::EVENT_WRITE == syscall::EVENT_WRITE { + loop { + let stream = match syscall::dup(server.as_raw_fd() as usize, b"listen").map_err(from_syscall_error) { + Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => break, + stream => stream? + }; + let stream = unsafe { File::from_raw_fd(stream as RawFd) }; + + event_file.write(&syscall::Event { + id: stream.as_raw_fd() as usize, + data: next_id, + flags: syscall::EVENT_READ | syscall::EVENT_WRITE, + })?; + + clients.insert(next_id, stream); + println!("-> Spawned client #{}", next_id); + next_id += 1; + } + } + } else { + println!("Client #{} received flags: {:?}", event.data, event.flags); + let client = clients.get_mut(&event.data).unwrap(); + + if event.flags & syscall::EVENT_READ == syscall::EVENT_READ { + println!("-> Reading"); + let mut buf = [0; 128]; + loop { + let len = match client.read(&mut buf) { + Ok(0) => { + println!("--> EOF"); + clients.remove(&event.data); + continue 'outer; + }, + Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => break, + len => len? + }; + println!("--> Read {}/128 bytes: {:?}", len, str::from_utf8(&buf[..len])); + } + } + if event.flags & syscall::EVENT_WRITE == syscall::EVENT_WRITE { + println!("-> Writing"); + const BUF: &str = "Hello from the log server\n"; + let mut written = 0; + while written < BUF.len() { + let len = match client.write(BUF[written..].as_bytes()) { + Ok(0) => panic!("EOF should never happen here"), + Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => break, + len => len? + }; + println!("--> Wrote {}/{} bytes: {:?}", len, BUF.len(), &BUF[written..]); + written += len; + } + } + } + } +} diff --git a/ipcd/examples/main.rs b/ipcd/examples/main.rs new file mode 100644 index 0000000000..7a15762386 --- /dev/null +++ b/ipcd/examples/main.rs @@ -0,0 +1,113 @@ +use std::{ + fs::{File, OpenOptions}, + io::{self, prelude::*}, + os::unix::io::{AsRawFd, FromRawFd, RawFd}, + thread, + time::Duration +}; + +fn from_syscall_error(error: syscall::Error) -> io::Error { + io::Error::from_raw_os_error(error.errno as i32) +} +fn nonblock(file: &File) -> io::Result<()> { + syscall::fcntl(file.as_raw_fd() as usize, syscall::F_SETFL, syscall::O_NONBLOCK) + .map(|_| ()) + .map_err(from_syscall_error) +} +fn dup(file: &File, buf: &str) -> io::Result { + let stream = syscall::dup(file.as_raw_fd() as usize, buf.as_bytes()).map_err(from_syscall_error)?; + Ok(unsafe { File::from_raw_fd(stream as RawFd) }) +} + +fn main() -> io::Result<()> { + let mut buf = [0; 5]; + let server = File::create("chan:hello_world")?; + { + println!("Testing O_EXCL..."); + assert_eq!( + OpenOptions::new() + .write(true) + .create_new(true) + .open("chan:hello_world").unwrap_err().kind(), + io::ErrorKind::AlreadyExists + ); + + println!("Testing connecting..."); + + File::open("chan:hello_world")?; // closed connection will silently be skipped + let mut client = File::create("chan:hello_world")?; // O_CREAT without O_EXCL does nothing + let tmp = File::open("chan:hello_world")?; // multiple connections are handled + + let mut stream = dup(&server, "listen")?; + assert!(dup(&server, "listen").is_ok()); + drop(tmp); + + println!("Testing basic I/O..."); + + stream.write(b"abc")?; + stream.flush()?; + println!("-> Wrote message"); + + assert_eq!(client.read(&mut buf)?, 3); + assert_eq!(&buf[..3], b"abc"); + println!("-> Read message"); + + println!("Testing close..."); + + drop(client); + assert_eq!(stream.write(b"a").unwrap_err().kind(), io::ErrorKind::BrokenPipe); + assert_eq!(stream.read(&mut buf)?, 0); + } + println!("Testing alternative connect method..."); + + let mut client = dup(&server, "connect")?; + let mut stream = dup(&server, "listen")?; + + println!("Testing blocking I/O..."); + + let mut client_clone = client.try_clone()?; + + let thread = thread::spawn(move || -> io::Result<()> { + println!("--> Thread: Sleeping for 1 second..."); + thread::sleep(Duration::from_secs(1)); + println!("--> Thread: Writing..."); + client_clone.write(b"def")?; + client_clone.flush()?; + Ok(()) + }); + + assert_eq!(stream.read(&mut buf)?, 3); + assert_eq!(&buf[..3], b"def"); + println!("-> Read message"); + + thread.join().unwrap().unwrap(); + + println!("Testing non-blocking I/O..."); + + nonblock(&client)?; + nonblock(&server)?; + + assert_eq!(client.read(&mut buf).unwrap_err().kind(), io::ErrorKind::WouldBlock); + println!("-> Read would block"); + + assert_eq!(dup(&server, "listen").unwrap_err().kind(), io::ErrorKind::WouldBlock); + println!("-> Accept would block"); + + drop(client); + { + let mut client = File::open("chan:hello_world")?; + nonblock(&client)?; + + assert_eq!(client.write(b"a").unwrap_err().kind(), io::ErrorKind::WouldBlock); + println!("-> Write before accept would block"); + } + + let mut client = dup(&server, "connect")?; + nonblock(&client)?; + + assert_eq!(client.write(b"a").unwrap_err().kind(), io::ErrorKind::WouldBlock); + println!("-> Write before accept would block (alternative connection method)"); + + println!("Everything tested!"); + Ok(()) +} diff --git a/ipcd/examples/shm.rs b/ipcd/examples/shm.rs new file mode 100644 index 0000000000..51cd984716 --- /dev/null +++ b/ipcd/examples/shm.rs @@ -0,0 +1,53 @@ +use std::{ + fs::File, + io, + os::unix::io::AsRawFd, + slice +}; + +fn from_syscall_error(error: syscall::Error) -> io::Error { + io::Error::from_raw_os_error(error.errno as i32) +} +fn main() -> Result<(), io::Error> { + let file1 = File::open("shm:example")?; + let file2 = File::open("shm:example")?; + + let one = unsafe { + slice::from_raw_parts_mut( + syscall::fmap(file1.as_raw_fd() as usize, &syscall::Map { + offset: 0, + size: 128, + flags: syscall::PROT_READ | syscall::PROT_WRITE | syscall::MAP_SHARED, + address: 0, + }).map_err(from_syscall_error)? as *mut u8, + 128 + ) + }; + // FIXME: While the length can be unaligned, the offset cannot. This test is incorrectly + // written. + let two = unsafe { + slice::from_raw_parts_mut( + syscall::fmap(file2.as_raw_fd() as usize, &syscall::Map { + offset: 64, + size: 64, + flags: syscall::PROT_READ | syscall::PROT_WRITE | syscall::MAP_SHARED, + address: 0, + }).map_err(from_syscall_error)? as *mut u8, + 64 + ) + }; + + println!("Testing writing between"); + for i in 0..128 { + one[i as usize] = i; + } + for i in 0..64 { + assert_eq!(two[i as usize], 64 + i); + } + + println!("Testing fpath"); + let mut buf = [0; 128]; + let len = syscall::fpath(file1.as_raw_fd() as usize, &mut buf).map_err(from_syscall_error)?; + assert_eq!(&buf[..len], b"shm:example"); + Ok(()) +} diff --git a/ipcd/examples/shm_demo.rs b/ipcd/examples/shm_demo.rs new file mode 100644 index 0000000000..b3f47515c4 --- /dev/null +++ b/ipcd/examples/shm_demo.rs @@ -0,0 +1,30 @@ +use std::{ + fs::File, + io, + mem, + os::unix::io::AsRawFd, + thread, + time::Duration +}; + +fn from_syscall_error(error: syscall::Error) -> io::Error { + io::Error::from_raw_os_error(error.errno as i32) +} +fn main() -> Result<(), io::Error> { + let file = File::open("shm:counter")?; + println!("Reading from map... "); + let counter = unsafe { + &mut *(syscall::fmap(file.as_raw_fd() as usize, &syscall::Map { + offset: 0, + address: 0, + size: mem::size_of::(), + flags: syscall::PROT_READ | syscall::PROT_WRITE | syscall::MAP_SHARED, + }).map_err(from_syscall_error)? as *mut usize) + }; + println!("Read value {}", counter); + *counter += 1; + println!("Increased value to {}", counter); + + thread::sleep(Duration::from_secs(1)); + Ok(()) +} diff --git a/ipcd/examples/unnamed.rs b/ipcd/examples/unnamed.rs new file mode 100644 index 0000000000..914bc930fc --- /dev/null +++ b/ipcd/examples/unnamed.rs @@ -0,0 +1,37 @@ +use std::{ + fs::File, + io::{self, prelude::*}, + os::unix::io::{AsRawFd, FromRawFd, RawFd} +}; + +fn from_syscall_error(error: syscall::Error) -> io::Error { + io::Error::from_raw_os_error(error.errno as i32) +} +fn dup(file: &File, buf: &str) -> io::Result { + let stream = syscall::dup(file.as_raw_fd() as usize, buf.as_bytes()).map_err(from_syscall_error)?; + Ok(unsafe { File::from_raw_fd(stream as RawFd) }) +} + +fn main() -> io::Result<()> { + let mut buf = [0; 5]; + let server = File::create("chan:")?; + + let mut client = dup(&server, "connect")?; + let mut stream = dup(&server, "listen")?; + + println!("Testing basic I/O..."); + + stream.write(b"abc")?; + stream.flush()?; + println!("-> Wrote message"); + + assert_eq!(client.read(&mut buf)?, 3); + assert_eq!(&buf[..3], b"abc"); + println!("-> Read message"); + + println!("Testing connecting to unnamed socket by name (makes no sense)..."); + assert_eq!(File::open("chan:").unwrap_err().kind(), io::ErrorKind::NotFound); + + println!("Everything tested!"); + Ok(()) +} diff --git a/ipcd/redoxer.sh b/ipcd/redoxer.sh new file mode 100755 index 0000000000..2bcdc680bd --- /dev/null +++ b/ipcd/redoxer.sh @@ -0,0 +1,20 @@ +#!/bin/sh + +set -e + +old_path="file:/bin/ipcd" +new_path="target/x86_64-unknown-redox/debug/ipcd" +if [ -e "${new_path}" ] +then + mv -v "${new_path}" "${old_path}" + shutdown --reboot +fi + +while [ "$#" != "0" ] +do + example="$1" + shift + + echo "# ${example} #" + "target/x86_64-unknown-redox/debug/examples/${example}" +done diff --git a/ipcd/src/chan.rs b/ipcd/src/chan.rs new file mode 100644 index 0000000000..087bbca90a --- /dev/null +++ b/ipcd/src/chan.rs @@ -0,0 +1,359 @@ +use std::{ + cmp, + collections::{HashMap, VecDeque}, +}; +use syscall::{flag::*, error::*, Error}; +use redox_scheme::{SchemeBlockMut, V2}; + +#[derive(Debug, Default)] +pub struct Client { + buffer: Vec, + remote: Connection +} +#[derive(Debug, Default)] +pub struct Listener { + path: Option, + awaiting: VecDeque +} +#[derive(Debug)] +pub enum Extra { + Client(Client), + Listener(Listener) +} +impl Default for Extra { + fn default() -> Self { + Extra::Client(Client::default()) + } +} +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum Connection { + Waiting, + Open(usize), + Closed +} +impl Default for Connection { + fn default() -> Self { + Connection::Waiting + } +} + +#[derive(Debug, Default)] +pub struct Handle { + flags: usize, + extra: Extra, + path: Option, +} +impl Handle { + /// Duplicate this listener handle into one that is linked to the + /// specified remote. + /// Does NOT error if this is not a listener + pub fn accept(&self, remote: usize) -> Self { + Self { + flags: self.flags, + extra: Extra::Client(Client { + remote: Connection::Open(remote), + ..Client::default() + }), + ..Default::default() + } + } + + /// Mark this listener handle as having a connection which can be + /// accepted, but only if it is ready to accept. + /// Errors if this is not a listener + pub fn connect(&mut self, other: usize) -> Result<()> { + match self.extra { + Extra::Listener(ref mut listener) => { + listener.awaiting.push_back(other); + Ok(()) + }, + _ => Err(Error::new(EBADF)) + } + } + + /// Error if this is not a listener + pub fn require_listener(&mut self) -> Result<&mut Listener> { + match self.extra { + Extra::Listener(ref mut listener) => Ok(listener), + _ => Err(Error::new(EBADF)) + } + } + + /// Error if this is not a client + pub fn require_client(&mut self) -> Result<&mut Client> { + match self.extra { + Extra::Client(ref mut client) => Ok(client), + _ => Err(Error::new(EBADF)) + } + } +} + +pub struct ChanScheme { + handles: HashMap, + listeners: HashMap, + next_id: usize, + pub socket: redox_scheme::Socket, +} +impl ChanScheme { + pub fn new() -> Result { + Ok(Self { + handles: HashMap::new(), + listeners: HashMap::new(), + next_id: 0, + socket: redox_scheme::Socket::::nonblock("chan")?, + }) + } +} + +impl SchemeBlockMut for ChanScheme { + // ___ ____ _____ _ _ + // / _ \| _ \| ____| \ | | + // | | | | |_) | _| | \| | + // | |_| | __/| |___| |\ | + // \___/|_| |_____|_| \_| + + fn open(&mut self, path: &str, flags: usize, _uid: u32, _gid: u32) -> Result> { + let new_id = self.next_id; + let mut new = Handle::default(); + new.flags = flags; + + let create = flags & O_CREAT == O_CREAT; + + if create && !self.listeners.contains_key(path) { + let mut listener = Listener::default(); + if !path.is_empty() { + self.listeners.insert(String::from(path), new_id); + listener.path = Some(String::from(path)); + } + new.extra = Extra::Listener(listener); + } else if create && flags & O_EXCL == O_EXCL { + return Err(Error::new(EEXIST)); + } else { + // Connect to existing if: O_CREAT isn't set or it already exists + // and O_EXCL isn't set + let listener_id = *self.listeners.get(path).ok_or(Error::new(ENOENT))?; + let listener = self.handles.get_mut(&listener_id).expect("orphan listener left over"); + listener.connect(new_id)?; + + // smoltcp sends writeable whenever a listener gets a + // client, we'll do the same too (but also readable, why + // not) + self.socket.post_fevent(listener_id, (EVENT_READ | EVENT_WRITE).bits())?; + } + + self.handles.insert(new_id, new); + self.next_id += 1; + Ok(Some(new_id)) + } + fn dup(&mut self, id: usize, buf: &[u8]) -> Result> { + match buf { + b"listen" => { + loop { + let handle = self.handles.get_mut(&id).ok_or(Error::new(EBADF))?; + let listener = handle.require_listener()?; + let listener_path = listener.path.clone(); + + break if let Some(remote_id) = listener.awaiting.pop_front() { + let new_id = self.next_id; + let mut new = handle.accept(remote_id); + + // Hook the remote side, assuming it's still + // connected, up to this one so the connection is + // mutal. + let remote = match self.handles.get_mut(&remote_id) { + Some(client) => client, + None => continue // Check next client + }; + match remote.extra { + Extra::Client(ref mut client) => { + client.remote = Connection::Open(new_id); + }, + Extra::Listener(_) => panic!("newly created handle can't possibly be a listener") + } + self.socket.post_fevent(remote_id, EVENT_WRITE.bits())?; + + new.path = listener_path; + + self.handles.insert(new_id, new); + self.next_id += 1; + Ok(Some(new_id)) + } else if handle.flags & O_NONBLOCK == O_NONBLOCK { + Err(Error::new(EAGAIN)) + } else { + Ok(None) + }; + } + }, + b"connect" => { + let new_id = self.next_id; + let new = Handle::default(); + + let handle = self.handles.get_mut(&id).ok_or(Error::new(EBADF))?; + handle.require_listener()?; + handle.connect(new_id)?; + + // smoltcp sends writeable whenever a listener gets a + // client, we'll do the same too (but also readable, + // why not) + self.socket.post_fevent(id, (EVENT_READ | EVENT_WRITE).bits())?; + + self.handles.insert(new_id, new); + self.next_id += 1; + Ok(Some(new_id)) + }, + _ => { + // If a buf is provided, different than "connect" / "listen", + // turn the socket into a named socket. + + if buf.is_empty() { + return Err(Error::new(EBADF)); + } + + let path = core::str::from_utf8(buf).map_err(|_| Error::new(EBADF))?; + + let handle = self.handles.get_mut(&id).ok_or(Error::new(EBADF))?; + if handle.path.is_some() { + return Err(Error::new(EBADF)); + } + + let flags = handle.flags; + return self.open(path, flags, 0, 0); + } + } + } + + // ___ ___ ___ ____ _ ___ ____ _____ + // |_ _/ _ \ ( _ ) / ___| | / _ \/ ___|| ____| + // | | | | | / _ \/\ | | | | | | | \___ \| _| + // | | |_| | | (_> < | |___| |__| |_| |___) | |___ + // |___\___/ \___/\/ \____|_____\___/|____/|_____| + + fn write(&mut self, id: usize, buf: &[u8], _offset: u64, flags: u32) -> Result> { + let handle = self.handles.get_mut(&id).ok_or(Error::new(EBADF))?; + let client = handle.require_client()?; + + if let Connection::Open(remote_id) = client.remote { + let remote = self.handles.get_mut(&remote_id).unwrap(); + match remote.extra { + Extra::Client(ref mut client) => { + client.buffer.extend(buf); + if client.buffer.len() == buf.len() { + // Send readable only if it wasn't readable + // before + self.socket.post_fevent(remote_id, EVENT_READ.bits())?; + } + Ok(Some(buf.len())) + }, + Extra::Listener(_) => panic!("somehow, a client was connected to a listener directly") + } + } else if client.remote == Connection::Closed { + Err(Error::new(EPIPE)) + } else if (flags as usize) & O_NONBLOCK == O_NONBLOCK { + Err(Error::new(EAGAIN)) + } else { + Ok(None) + } + } + fn fpath(&mut self, id: usize, buf: &mut [u8]) -> Result> { + // Write scheme name + const PREFIX: &[u8] = b"chan:"; + let len = cmp::min(PREFIX.len(), buf.len()); + buf[..len].copy_from_slice(&PREFIX[..len]); + if len < PREFIX.len() { + return Ok(Some(len)); + } + + // Write path + let handle = self.handles.get(&id).ok_or(Error::new(EBADF))?; + let path = handle.path.as_ref().ok_or(Error::new(EBADF))?; + let len = cmp::min(path.len(), buf.len() - PREFIX.len()); + buf[PREFIX.len()..][..len].copy_from_slice(&path.as_bytes()[..len]); + + Ok(Some(PREFIX.len() + len)) + } + fn fsync(&mut self, id: usize) -> Result> { + self.handles.get(&id) + .ok_or(Error::new(EBADF)) + .and(Ok(Some(id))) + } + fn read(&mut self, id: usize, buf: &mut [u8], _offset: u64, flags: u32) -> Result> { + let handle = self.handles.get_mut(&id).ok_or(Error::new(EBADF))?; + let client = handle.require_client()?; + + if !client.buffer.is_empty() { + let len = cmp::min(buf.len(), client.buffer.len()); + buf[..len].copy_from_slice(&client.buffer[..len]); + client.buffer.drain(..len); + Ok(Some(len)) + } else if client.remote == Connection::Closed { + // Remote dropped, send EOF + Ok(Some(0)) + } else if (flags as usize) & O_NONBLOCK == O_NONBLOCK { + Err(Error::new(EAGAIN)) + } else { + Ok(None) + } + } + fn close(&mut self, id: usize) -> Result> { + let handle = self.handles.remove(&id).ok_or(Error::new(EBADF))?; + + match handle.extra { + Extra::Client(client) => if let Connection::Open(remote_id) = client.remote { + let remote = self.handles.get_mut(&remote_id).unwrap(); + + match remote.extra { + Extra::Client(ref mut client) => { + client.remote = Connection::Closed; + if client.buffer.is_empty() { + // Post readable on EOF only if it wasn't + // readable before + self.socket.post_fevent(remote_id, EVENT_READ.bits())?; + } + }, + Extra::Listener(_) => panic!("a client can't be connected to a listener!") + } + }, + Extra::Listener(listener) => if let Some(path) = listener.path { + self.listeners.remove(&path); + } + } + Ok(Some(0)) + } + + + // ____ _ ____ _ __ __ _____ _____ _____ ____ ____ + // | _ \ / \ | _ \ / \ | \/ | ____|_ _| ____| _ \/ ___| + // | |_) / _ \ | |_) | / _ \ | |\/| | _| | | | _| | |_) \___ \ + // | __/ ___ \| _ < / ___ \| | | | |___ | | | |___| _ < ___) | + // |_| /_/ \_\_| \_\/_/ \_\_| |_|_____| |_| |_____|_| \_\____/ + + fn fcntl(&mut self, id: usize, cmd: usize, arg: usize) -> Result> { + let handle = self.handles.get_mut(&id).ok_or(Error::new(EBADF))?; + match cmd { + F_GETFL => Ok(Some(handle.flags)), + F_SETFL => { + handle.flags = arg; + Ok(Some(0)) + }, + _ => Err(Error::new(EINVAL)) + } + } + fn fevent(&mut self, id: usize, _flags: EventFlags) -> Result> { + let handle = self.handles.get_mut(&id).ok_or(Error::new(EBADF))?; + let mut events = EventFlags::empty(); + match handle.extra { + Extra::Client(ref client) => { + if let Connection::Open(_) = client.remote { + events |= EVENT_WRITE; + } + if !client.buffer.is_empty() || client.remote == Connection::Closed { + events |= EVENT_READ; + } + }, + Extra::Listener(ref listener) => if !listener.awaiting.is_empty() { + events |= EVENT_READ | EVENT_WRITE; + } + } + Ok(Some(events)) + } +} diff --git a/ipcd/src/main.rs b/ipcd/src/main.rs new file mode 100644 index 0000000000..710a784439 --- /dev/null +++ b/ipcd/src/main.rs @@ -0,0 +1,158 @@ +#![feature(int_roundings, let_chains)] + +use std::collections::VecDeque; +use event::{EventQueue, EventFlags}; +use redox_scheme::{CallRequest, RequestKind, Response, SignalBehavior}; +use syscall::{Error, Result, EAGAIN, EWOULDBLOCK, ENODEV, EINTR}; + +mod chan; +mod shm; + +use self::chan::ChanScheme; +use self::shm::ShmScheme; + +fn main() { + redox_daemon::Daemon::new(move |daemon| { + // TODO: Better error handling + match inner(daemon) { + Ok(()) => std::process::exit(0), + Err(error) => { + println!("ipcd failed: {error}"); + std::process::exit(1); + } + } + }).expect("ipcd: failed to daemonize"); +} + +fn inner(daemon: redox_daemon::Daemon) -> Result<()> { + event::user_data! { + enum EventSource { + ChanSocket, + ShmSocket, + } + } + let chan = ChanScheme::new()?; + let shm = ShmScheme::new()?; + daemon.ready().unwrap(); + + // Create event listener for both files + let mut event_queue = EventQueue::::new()?; + + event_queue.subscribe(chan.socket.inner().raw(), EventSource::ChanSocket, EventFlags::READ)?; + event_queue.subscribe(shm.socket.inner().raw(), EventSource::ShmSocket, EventFlags::READ)?; + + struct Todo { + req: Option, + canceling: bool, + } + let mut todo = VecDeque::::with_capacity(16); + + libredox::call::setrens(0, 0)?; + + let mut chan_opt = Some(chan); + let mut shm_opt = Some(shm); + while chan_opt.is_some() || shm_opt.is_some() { + let Some(event_res) = event_queue.next() else { + break; + }; + let event = event_res?; + + match event.user_data { + EventSource::ChanSocket => { + let mut error: Option = None; + + let unmount = if let Some(ref mut chan) = chan_opt { + let eof = loop { + match chan.socket.next_request(SignalBehavior::Restart) { + Ok(None) => break true, + Ok(Some(request)) => match request.kind() { + RequestKind::Call(request) => todo.push_front(Todo { req: Some(request), canceling: false }), + RequestKind::Cancellation(request) => { + if let Some(affected_packet) = todo.iter_mut().find(|t| t.req.as_ref().map_or(false, |r| r.request().request_id() == request.id)) { + affected_packet.canceling = true; + } + } + _ => (), + } + Err(Error { errno: EAGAIN | EWOULDBLOCK }) => break false, + Err(error) => return Err(error), + } + }; + + // Process queue, delete finished items + todo.retain_mut(|slot| { + let req = slot.req.take().unwrap(); + + match req.handle_scheme_block_mut(chan) { + Some(res) => { + if let Err(err) = chan.socket.write_response(res, SignalBehavior::Restart) { + error = Some(err); + } + false + } + None if slot.canceling => { + if let Err(err) = chan.socket.write_response(Response::new(&req, Err(Error::new(EINTR))), SignalBehavior::Restart) { + error = Some(err); + } + false + } + None => { + slot.req = Some(req); + true + } + } + }); + + eof + } else { + false + }; + + if unmount && let Some(chan) = chan_opt.take() { + for slot in todo.drain(..) { + let res = if slot.canceling { + Err(Error::new(EINTR)) + } else { + Err(Error::new(ENODEV)) + }; + if let Err(err) = chan.socket.write_response(Response::new(&slot.req.unwrap(), res), SignalBehavior::Restart) { + error = Some(err); + } + } + } + + if let Some(err) = error { + return Err(err); + } + }, + EventSource::ShmSocket => { + let unmount = if let Some(ref mut shm) = shm_opt { + let eof = loop { + match shm.socket.next_request(SignalBehavior::Restart) { + Ok(None) => break true, + Ok(Some(request)) => match request.kind() { + RequestKind::Call(request) => { + let response = request.handle_scheme_mut(shm); + shm.socket.write_response(response, SignalBehavior::Restart)?; + } + _ => (), + }, + Err(Error { errno: EAGAIN | EWOULDBLOCK }) => break false, + Err(err) => return Err(err), + }; + }; + + eof + } else { + false + }; + + if unmount { + shm_opt.take(); + } + } + } + } + + Ok(()) +} diff --git a/ipcd/src/shm.rs b/ipcd/src/shm.rs new file mode 100644 index 0000000000..ac2b87ff0b --- /dev/null +++ b/ipcd/src/shm.rs @@ -0,0 +1,117 @@ +use std::{ + cmp, + collections::{HashMap, hash_map::Entry}, + rc::Rc, +}; +use syscall::{error::*, Error, Map, Result, MapFlags, PAGE_SIZE, MAP_PRIVATE}; +use redox_scheme::{SchemeMut, V2}; + +#[derive(Default)] +pub struct ShmHandle { + buffer: Option, + refs: usize +} +pub struct ShmScheme { + maps: HashMap, ShmHandle>, + handles: HashMap>, + next_id: usize, + pub socket: redox_scheme::Socket, +} +impl ShmScheme { + pub fn new() -> Result { + Ok(Self { + maps: HashMap::new(), + handles: HashMap::new(), + next_id: 0, + socket: redox_scheme::Socket::::nonblock("shm")?, + }) + } +} + +impl SchemeMut for ShmScheme { + fn open(&mut self, path: &str, _flags: usize, _uid: u32, _gid: u32) -> Result { + let path = Rc::from(path); + let entry = self.maps.entry(Rc::clone(&path)).or_insert(ShmHandle::default()); + entry.refs += 1; + self.handles.insert(self.next_id, path); + + let id = self.next_id; + self.next_id += 1; + Ok(id) + } + fn fpath(&mut self, id: usize, buf: &mut [u8]) -> Result { + // Write scheme name + const PREFIX: &[u8] = b"shm:"; + let len = cmp::min(PREFIX.len(), buf.len()); + buf[..len].copy_from_slice(&PREFIX[..len]); + if len < PREFIX.len() { + return Ok(len); + } + + // Write path + let path = self.handles.get(&id).ok_or(Error::new(EBADF))?; + let len = cmp::min(path.len(), buf.len() - PREFIX.len()); + buf[PREFIX.len()..][..len].copy_from_slice(&path.as_bytes()[..len]); + + Ok(PREFIX.len() + len) + } + fn close(&mut self, id: usize) -> Result { + let path = self.handles.remove(&id).ok_or(Error::new(EBADF))?; + let mut entry = match self.maps.entry(path) { + Entry::Occupied(entry) => entry, + Entry::Vacant(_) => panic!("handle pointing to nothing") + }; + entry.get_mut().refs -= 1; + if entry.get().refs == 0 { + // There is no other reference to this entry, drop + entry.remove_entry(); + } + Ok(0) + } + fn mmap_prep(&mut self, id: usize, offset: u64, size: usize, _: MapFlags) -> Result { + let path = self.handles.get(&id).ok_or(Error::new(EBADF))?; + let total_size = offset as usize + size; + match self.maps.get_mut(path).expect("handle pointing to nothing").buffer { + Some(ref mut buf) => { + if total_size > buf.len() { + return Err(Error::new(ERANGE)); + } + Ok(buf.as_ptr() + offset as usize) + }, + ref mut buf @ None => { + *buf = Some(MmapGuard::alloc(size.div_ceil(PAGE_SIZE))?); + Ok(buf.as_mut().unwrap().as_ptr() + offset as usize) + } + } + } +} + +pub struct MmapGuard { + base: usize, + size: usize, +} +impl MmapGuard { + pub fn alloc(page_count: usize) -> Result { + let size = page_count * PAGE_SIZE; + let base = unsafe { syscall::fmap(!0, &Map { offset: 0, size, flags: MAP_PRIVATE, address: 0 }) }?; + + Ok(Self { + base, + size, + }) + } + pub fn len(&self) -> usize { + self.size + } + pub fn as_ptr(&self) -> usize { + self.base + } +} +impl Drop for MmapGuard { + fn drop(&mut self) { + if self.size == 0 { + return; + } + let _ = unsafe { syscall::funmap(self.base, self.size) }; + } +}