openethereum/ipc/nano/src/lib.rs

281 lines
7.9 KiB
Rust
Raw Normal View History

2016-04-03 20:43:35 +02:00
// Copyright 2015, 2016 Ethcore (UK) Ltd.
// This file is part of Parity.
// Parity is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// Parity is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
2016-04-03 22:39:49 +02:00
//! IPC over nanomsg transport
2016-04-03 20:43:35 +02:00
extern crate ethcore_ipc as ipc;
extern crate nanomsg;
2016-04-03 23:00:57 +02:00
#[macro_use] extern crate log;
2016-04-03 22:39:49 +02:00
2016-04-12 09:18:39 +02:00
pub use ipc::{WithSocket, IpcInterface, IpcConfig};
2016-04-03 22:39:49 +02:00
use std::sync::*;
2016-04-05 23:10:24 +02:00
use nanomsg::{Socket, Protocol, Error, Endpoint, PollRequest, PollFd, PollInOut};
2016-04-12 09:18:39 +02:00
use std::ops::Deref;
2016-04-05 23:10:24 +02:00
const POLL_TIMEOUT: isize = 100;
2016-04-03 22:39:49 +02:00
2016-04-12 10:53:41 +02:00
/// Generic worker to handle service (binded) sockets
2016-04-03 22:39:49 +02:00
pub struct Worker<S> where S: IpcInterface<S> {
service: Arc<S>,
2016-04-04 19:47:16 +02:00
sockets: Vec<(Socket, Endpoint)>,
2016-04-05 23:10:24 +02:00
polls: Vec<PollFd>,
buf: Vec<u8>,
2016-04-03 22:39:49 +02:00
}
2016-04-12 10:53:41 +02:00
/// struct for guarding `_endpoint` (so that it wont drop)
/// derefs to client `S`
2016-04-12 09:18:39 +02:00
pub struct GuardedSocket<S> where S: WithSocket<Socket> {
client: Arc<S>,
_endpoint: Endpoint,
}
impl<S> Deref for GuardedSocket<S> where S: WithSocket<Socket> {
type Target = S;
fn deref(&self) -> &S {
&self.client
}
}
2016-04-12 10:53:41 +02:00
/// Spawns client <`S`> over specified address
/// creates socket and connects endpoint to it
/// for duplex (paired) connections with the service
2016-04-12 09:18:39 +02:00
pub fn init_client<S>(socket_addr: &str) -> Result<GuardedSocket<S>, SocketError> where S: WithSocket<Socket> {
let mut socket = try!(Socket::new(Protocol::Pair).map_err(|e| {
warn!(target: "ipc", "Failed to create ipc socket: {:?}", e);
SocketError::DuplexLink
}));
2016-04-12 10:34:56 +02:00
let endpoint = try!(socket.connect(socket_addr).map_err(|e| {
2016-04-12 09:18:39 +02:00
warn!(target: "ipc", "Failed to bind socket to address '{}': {:?}", socket_addr, e);
SocketError::DuplexLink
}));
Ok(GuardedSocket {
client: Arc::new(S::init(socket)),
_endpoint: endpoint,
})
}
2016-04-12 10:53:41 +02:00
/// Error occured while establising socket or endpoint
2016-04-03 23:54:30 +02:00
#[derive(Debug)]
2016-04-03 23:33:30 +02:00
pub enum SocketError {
2016-04-12 10:53:41 +02:00
/// Error establising duplex (paired) socket and/or endpoint
2016-04-03 23:33:30 +02:00
DuplexLink
}
2016-04-03 22:39:49 +02:00
impl<S> Worker<S> where S: IpcInterface<S> {
2016-04-12 10:53:41 +02:00
/// New worker over specified `service`
2016-04-03 22:58:18 +02:00
pub fn new(service: Arc<S>) -> Worker<S> {
2016-04-03 22:39:49 +02:00
Worker::<S> {
service: service.clone(),
sockets: Vec::new(),
2016-04-05 23:10:24 +02:00
polls: Vec::new(),
buf: Vec::new(),
2016-04-03 22:39:49 +02:00
}
}
2016-04-12 10:53:41 +02:00
/// Polls all sockets, reads and dispatches method invocations
2016-04-03 22:58:18 +02:00
pub fn poll(&mut self) {
2016-04-05 23:10:24 +02:00
let mut request = PollRequest::new(&mut self.polls[..]);
let _result_guard = Socket::poll(&mut request, POLL_TIMEOUT);
for (fd_index, fd) in request.get_fds().iter().enumerate() {
if fd.can_read() {
let (ref mut socket, _) = self.sockets[fd_index];
unsafe { self.buf.set_len(0); }
match socket.nb_read_to_end(&mut self.buf) {
Ok(method_sign_len) => {
if method_sign_len >= 2 {
2016-04-12 10:13:27 +02:00
2016-04-05 23:10:24 +02:00
// method_num
let method_num = self.buf[1] as u16 * 256 + self.buf[0] as u16;
// payload
let payload = &self.buf[2..];
// dispatching for ipc interface
let result = self.service.dispatch_buf(method_num, payload);
if let Err(e) = socket.nb_write(&result) {
warn!(target: "ipc", "Failed to write response: {:?}", e);
}
2016-04-04 00:44:30 +02:00
}
2016-04-05 23:10:24 +02:00
else {
warn!(target: "ipc", "Failed to read method signature from socket: unexpected message length({})", method_sign_len);
}
},
Err(Error::TryAgain) => {
},
Err(x) => {
warn!(target: "ipc", "Error polling connections {:?}", x);
panic!();
2016-04-03 23:00:57 +02:00
}
2016-04-03 23:33:30 +02:00
}
2016-04-03 22:58:18 +02:00
}
}
2016-04-03 22:39:49 +02:00
}
2016-04-03 23:33:30 +02:00
2016-04-12 10:53:41 +02:00
/// Stores nanomsg poll request for reuse
2016-04-05 23:10:24 +02:00
fn rebuild_poll_request(&mut self) {
self.polls = self.sockets.iter()
.map(|&(ref socket, _)| socket.new_pollfd(PollInOut::In))
.collect::<Vec<PollFd>>();
}
2016-04-12 10:53:41 +02:00
/// Add exclusive socket for paired client
/// Only one connection over this address is allowed
2016-04-03 23:33:30 +02:00
pub fn add_duplex(&mut self, addr: &str) -> Result<(), SocketError> {
let mut socket = try!(Socket::new(Protocol::Pair).map_err(|e| {
warn!(target: "ipc", "Failed to create ipc socket: {:?}", e);
SocketError::DuplexLink
2016-04-13 18:03:57 +02:00
}));
let endpoint = try!(socket.bind(addr).map_err(|e| {
warn!(target: "ipc", "Failed to bind socket to address '{}': {:?}", addr, e);
SocketError::DuplexLink
}));
self.sockets.push((socket, endpoint));
self.rebuild_poll_request();
Ok(())
}
/// Add generic socket for request-reply style communications
/// with multiple clients
pub fn add_reqrep(&mut self, addr: &str) -> Result<(), SocketError> {
let mut socket = try!(Socket::new(Protocol::Rep).map_err(|e| {
warn!(target: "ipc", "Failed to create ipc socket: {:?}", e);
SocketError::DuplexLink
2016-04-03 23:33:30 +02:00
}));
2016-04-04 19:47:16 +02:00
let endpoint = try!(socket.bind(addr).map_err(|e| {
2016-04-03 23:33:30 +02:00
warn!(target: "ipc", "Failed to bind socket to address '{}': {:?}", addr, e);
SocketError::DuplexLink
}));
2016-04-04 19:47:16 +02:00
self.sockets.push((socket, endpoint));
2016-04-05 23:10:24 +02:00
self.rebuild_poll_request();
2016-04-03 23:33:30 +02:00
Ok(())
}
}
#[cfg(test)]
2016-04-12 09:18:39 +02:00
mod service_tests {
2016-04-03 23:33:30 +02:00
use super::Worker;
use ipc::*;
2016-04-04 00:44:30 +02:00
use std::io::{Read, Write};
2016-04-03 23:54:30 +02:00
use std::sync::{Arc, RwLock};
2016-04-05 11:08:42 +02:00
use nanomsg::{Socket, Protocol, Endpoint};
2016-04-03 23:33:30 +02:00
2016-04-03 23:54:30 +02:00
struct TestInvoke {
method_num: u16,
params: Vec<u8>,
}
struct DummyService {
methods_stack: RwLock<Vec<TestInvoke>>,
}
impl DummyService {
fn new() -> DummyService {
DummyService { methods_stack: RwLock::new(Vec::new()) }
}
}
2016-04-03 23:33:30 +02:00
impl IpcInterface<DummyService> for DummyService {
2016-04-03 23:54:30 +02:00
fn dispatch<R>(&self, _r: &mut R) -> Vec<u8> where R: Read {
2016-04-03 23:33:30 +02:00
vec![]
}
2016-04-05 11:08:42 +02:00
fn dispatch_buf(&self, method_num: u16, buf: &[u8]) -> Vec<u8> {
2016-04-03 23:54:30 +02:00
self.methods_stack.write().unwrap().push(
TestInvoke {
method_num: method_num,
2016-04-05 11:08:42 +02:00
params: buf.to_vec(),
2016-04-03 23:54:30 +02:00
});
2016-04-03 23:33:30 +02:00
vec![]
}
}
2016-04-12 09:18:39 +02:00
impl IpcConfig for DummyService {}
2016-04-05 11:08:42 +02:00
fn dummy_write(addr: &str, buf: &[u8]) -> (Socket, Endpoint) {
2016-04-04 00:44:30 +02:00
let mut socket = Socket::new(Protocol::Pair).unwrap();
2016-04-04 19:47:16 +02:00
let endpoint = socket.connect(addr).unwrap();
2016-04-05 11:08:42 +02:00
socket.write(buf).unwrap();
(socket, endpoint)
2016-04-04 00:44:30 +02:00
}
2016-04-03 23:42:00 +02:00
#[test]
2016-04-03 23:33:30 +02:00
fn can_create_worker() {
2016-04-03 23:54:30 +02:00
let worker = Worker::<DummyService>::new(Arc::new(DummyService::new()));
2016-04-03 23:33:30 +02:00
assert_eq!(0, worker.sockets.len());
}
2016-04-03 23:54:30 +02:00
#[test]
fn can_add_duplex_socket_to_worker() {
let mut worker = Worker::<DummyService>::new(Arc::new(DummyService::new()));
2016-04-04 00:52:19 +02:00
worker.add_duplex("ipc:///tmp/parity-test10.ipc").unwrap();
2016-04-03 23:54:30 +02:00
assert_eq!(1, worker.sockets.len());
}
2016-04-04 00:44:30 +02:00
#[test]
fn worker_can_poll_empty() {
let service = Arc::new(DummyService::new());
let mut worker = Worker::<DummyService>::new(service.clone());
2016-04-04 00:52:19 +02:00
worker.add_duplex("ipc:///tmp/parity-test20.ipc").unwrap();
2016-04-04 00:44:30 +02:00
worker.poll();
assert_eq!(0, service.methods_stack.read().unwrap().len());
}
#[test]
fn worker_can_poll() {
2016-04-04 09:55:06 +02:00
let url = "ipc:///tmp/parity-test30.ipc";
let mut worker = Worker::<DummyService>::new(Arc::new(DummyService::new()));
worker.add_duplex(url).unwrap();
2016-04-04 00:44:30 +02:00
2016-04-05 11:08:42 +02:00
let (_socket, _endpoint) = dummy_write(url, &vec![0, 0, 7, 7, 6, 6]);
2016-04-05 23:10:24 +02:00
worker.poll();
2016-04-05 11:08:42 +02:00
assert_eq!(1, worker.service.methods_stack.read().unwrap().len());
assert_eq!(0, worker.service.methods_stack.read().unwrap()[0].method_num);
assert_eq!([7, 7, 6, 6], worker.service.methods_stack.read().unwrap()[0].params[..]);
}
#[test]
fn worker_can_poll_long() {
2016-04-05 11:11:05 +02:00
let url = "ipc:///tmp/parity-test40.ipc";
2016-04-05 11:08:42 +02:00
let mut worker = Worker::<DummyService>::new(Arc::new(DummyService::new()));
worker.add_duplex(url).unwrap();
let message = [0u8; 1024*1024];
let (_socket, _endpoint) = dummy_write(url, &message);
2016-04-05 23:10:24 +02:00
worker.poll();
2016-04-04 00:44:30 +02:00
2016-04-04 09:55:06 +02:00
assert_eq!(1, worker.service.methods_stack.read().unwrap().len());
2016-04-05 11:08:42 +02:00
assert_eq!(0, worker.service.methods_stack.read().unwrap()[0].method_num);
assert_eq!(vec![0u8; 1024*1024-2], worker.service.methods_stack.read().unwrap()[0].params);
2016-04-04 00:44:30 +02:00
}
2016-04-03 22:39:49 +02:00
}