aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--Cargo.toml1
-rwxr-xr-xexamples/einhorn_http.py24
-rw-r--r--src/main.rs48
3 files changed, 59 insertions, 14 deletions
diff --git a/Cargo.toml b/Cargo.toml
index 923ba92..4896865 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -14,3 +14,4 @@ chrono = "0.2"
chan = "0.1"
chan-signal = "0.1"
uuid = { version = "0.2", features = ["v4"] }
+# XXX: num = "0.1"
diff --git a/examples/einhorn_http.py b/examples/einhorn_http.py
index 7d93e61..304d01a 100755
--- a/examples/einhorn_http.py
+++ b/examples/einhorn_http.py
@@ -1,4 +1,11 @@
-#!/usr/bin/python3
+#!/usr/bin/env python3
+"""
+This small example program demonstrates one way to integerate with Einhorn using
+Python (3).
+
+It serves up the current working directory over HTTP on either the
+Einhorn-supplied socket or localhost:8080.
+"""
import os
import sys
@@ -16,11 +23,11 @@ class EinhornTCPServer(socketserver.TCPServer):
fd = int(os.environ['EINHORN_FD_0'])
print("Will try to listen with fd=%d" % fd)
except KeyError:
- print("Couldn't find EINHORN_FD_0 env variable... is this running under einhorn?")
- sys.exit(1)
+ raise EnvironmentError("Couldn't find EINHORN_FD_0 env variable... is this running under einhorn?")
- #self.socket = socket.fromfd(socket.AF_INET, socket.SOCK_STREAM, fd)
self.socket = socket.socket(fileno=fd)
+ # alternative?
+ #self.socket = socket.fromfd(socket.AF_INET, socket.SOCK_STREAM, fd)
try:
self.server_activate()
@@ -32,9 +39,14 @@ if __name__ == "__main__":
Handler = http.server.SimpleHTTPRequestHandler
try:
httpd = EinhornTCPServer(None, Handler)
- except:
+ except EnvironmentError as ee:
+ print(ee)
print("Falling back on vanilla http server on 8080")
httpd = socketserver.TCPServer(("localhost", 8080), Handler)
print("Serving!")
- httpd.serve_forever()
+ try:
+ httpd.serve_forever()
+ except KeyboardInterrupt:
+ print("Caught KeyboardInterrupt, shutting down")
+ httpd.server_close()
diff --git a/src/main.rs b/src/main.rs
index c1e06b7..e849779 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -87,7 +87,7 @@ impl Offspring {
}
pub fn spawn(&mut self, cfg: &mut EinConfig) -> Result<(), String> {
- if self.state != OffspringState::Expectant && self.state != OffspringState::Dead {
+ if self.is_active() {
return Err(format!("Can't spawn from state: {:?}", self.state));
}
self.process = Some(cfg.cmd.spawn().expect("error spawning"));
@@ -95,6 +95,34 @@ impl Offspring {
self.attempts = 0;
Ok(())
}
+
+ pub fn is_active(&self) -> bool {
+ match self.state {
+ OffspringState::Expectant => false,
+ OffspringState::Infancy => true,
+ OffspringState::Healthy => true,
+ OffspringState::Notified => true,
+ OffspringState::Dead => false,
+ }
+ }
+
+ pub fn signal(&mut self, sig: Signal) {
+ if !self.is_active() {
+ return;
+ }
+ let nix_sig = match sig {
+ Signal::HUP => nix::sys::signal::Signal::SIGHUP,
+ Signal::INT => nix::sys::signal::Signal::SIGINT,
+ _ => { println!("Unexpected signal: {:?}", sig); return; },
+ };
+ match self.process {
+ Some(ref p) => { nix::sys::signal::kill(p.id() as i32, nix_sig).unwrap(); },
+ None => (),
+ }
+ // TODO: abs probably isn't the best way to i32 -> u32 here
+ //nix::sys::signal::kill(num::abs(self.process.unwrap().id()), nix_sig);
+ //let pid = self.process.unwrap().id();
+ }
}
#[derive(Copy, Clone, Debug, PartialEq)]
@@ -107,7 +135,6 @@ fn shepard(mut cfg: EinConfig, signal_rx: Receiver<Signal>) {
//// create timer
let timer = timer::Timer::new();
- // XXX: these signatures are bogus
let (timer_tx, timer_rx): (Sender<TimerAction>, Receiver<TimerAction>) = chan::async();
//// birth the initial set of offspring
@@ -127,11 +154,15 @@ fn shepard(mut cfg: EinConfig, signal_rx: Receiver<Signal>) {
//// infinite select() loop over timers, signals, rpc
loop {
chan_select! {
- timer_rx.recv() => println!("Timer tick'd"),
- signal_rx.recv() -> sig => {
- println!("");
- println!("Signal received! {:?}", sig);
- break;
+ timer_rx.recv() => { println!("Timer tick'd"); "TIMER" },
+ signal_rx.recv() -> sig => match sig.expect("Error with signal handler") {
+ // XXX: Signal::HUP => brood.iter().for_each(|o| o.signal(sig)),
+ Signal::INT | Signal::TERM=> {
+ println!("Notifying children...");
+ // XXX: brood.iter().for_each(|o| o.signal(sig));
+ break;
+ },
+ _ => "Other"
}
}
}
@@ -144,8 +175,8 @@ fn shepard(mut cfg: EinConfig, signal_rx: Receiver<Signal>) {
None => (),
}
}
-*/
println!("Done.");
+*/
}
fn print_usage(opts: Options) {
@@ -209,6 +240,7 @@ fn main() {
builder.init().unwrap();
let binds: Vec<TcpListener> = sock_addrs.iter().map(|sa| {
+ // XXX: SO_REUSE here
TcpListener::bind(sa).unwrap()
}).collect();