Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

setting variable reconnect timeout #6

Merged
merged 2 commits into from
Jul 3, 2024
Merged

setting variable reconnect timeout #6

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
704993a
fix: setting variable reconnect timeout
ruslanti Jul 3, 2024
9104e8a
fix: moved reconnect timeout to Disconnect state
ruslanti Jul 3, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 26 additions & 14 deletions nginx_module/src/unix_socket.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@ use std::{
borrow::Cow, cell::RefCell, collections::VecDeque, marker::PhantomPinned, mem::MaybeUninit,
os::unix::ffi::OsStrExt, path::Path, pin::Pin,
};

use libc::{c_void, sockaddr_un};

use crate::{
Expand All @@ -28,6 +27,7 @@ enum State {
},
Disconnected {
event: Box<ngx_event_t>,
reconnect_timeout: usize
},
}

Expand Down Expand Up @@ -57,7 +57,8 @@ struct WriteBuffer {
end: u32,
}

const TIMEOUT_MS: usize = 500;
const MIN_TIMEOUT_MS: usize = 1;
const TIMEOUT_MS: usize = 255;

impl UnixSocket {
pub fn connect(
Expand All @@ -82,17 +83,17 @@ impl UnixSocket {
Box::new(MaybeUninit::zeroed().assume_init());
ev.handler = Some(on_reconnect_timeout);
ev.log = (*ngx_cycle).log;
ngx_event_add_timer(&mut *ev, TIMEOUT_MS);
State::Disconnected { event: ev }
ngx_event_add_timer(&mut *ev, MIN_TIMEOUT_MS);
State::Disconnected { event: ev, reconnect_timeout: MIN_TIMEOUT_MS }
},
}
}
None => unsafe {
let mut ev: Box<ngx_event_t> = Box::new(MaybeUninit::zeroed().assume_init());
ev.handler = Some(on_reconnect_timeout);
ev.log = (*ngx_cycle).log;
ngx_event_add_timer(&mut *ev, TIMEOUT_MS);
State::Disconnected { event: ev }
ngx_event_add_timer(&mut *ev, MIN_TIMEOUT_MS);
State::Disconnected { event: ev, reconnect_timeout: MIN_TIMEOUT_MS }
},
};

Expand Down Expand Up @@ -120,7 +121,7 @@ impl UnixSocket {
(*(**conn).write).handler = Some(on_write);
(*(**conn).read).handler = Some(on_read);
},
State::Disconnected { event } => {
State::Disconnected { event, .. } => {
event.data = (&*inner as *const Inner as *mut Inner).cast();
(inner.after_handshake.borrow_mut())(); // discard send data, it is disconnected
}
Expand All @@ -139,7 +140,7 @@ impl UnixSocket {
let mut dummy_ev: Box<ngx_event_t> = Box::new(MaybeUninit::zeroed().assume_init());
dummy_ev.handler = None;
dummy_ev.log = (*ngx_cycle).log;
*self.0.state.borrow_mut() = State::Disconnected { event: dummy_ev }
*self.0.state.borrow_mut() = State::Disconnected { event: dummy_ev, reconnect_timeout: MIN_TIMEOUT_MS}
}
}
}
Expand Down Expand Up @@ -221,7 +222,7 @@ impl Drop for State {
ngx_close_connection(*conn);
}
},
Self::Disconnected { event } => unsafe {
Self::Disconnected { event, .. } => unsafe {
if event.active() != 0 {
ngx_event_del_timer(event.as_mut());
}
Expand Down Expand Up @@ -278,6 +279,7 @@ impl Inner {
if let Err(Disconnected) = unsafe { buffers.send(*conn) } {
*state = State::Disconnected {
event: self.create_and_schedule_reconnect(),
reconnect_timeout: MIN_TIMEOUT_MS
};
}
}
Expand All @@ -290,7 +292,7 @@ impl Inner {
ev.log = (*ngx_cycle).log;
ev.data = (self as *const Self as *mut Self).cast();

ngx_event_add_timer(&mut *ev, TIMEOUT_MS);
ngx_event_add_timer(&mut *ev, 0);

ev
}
Expand Down Expand Up @@ -472,6 +474,7 @@ unsafe extern "C" fn on_read(rev: *mut ngx_event_t) {
Err(_) => {
*state = State::Disconnected {
event: (*data).create_and_schedule_reconnect(),
reconnect_timeout: MIN_TIMEOUT_MS,
}
}
}
Expand All @@ -488,6 +491,7 @@ unsafe extern "C" fn on_read(rev: *mut ngx_event_t) {
if !back_data.is_empty() && (*data).write(&back_data).is_err() {
*(*data).state.borrow_mut() = State::Disconnected {
event: (*data).create_and_schedule_reconnect(),
reconnect_timeout: MIN_TIMEOUT_MS,
};
}
}
Expand All @@ -497,26 +501,30 @@ unsafe extern "C" fn on_read(rev: *mut ngx_event_t) {
if !back_data.is_empty() && (*data).write(&back_data).is_err() {
*(*data).state.borrow_mut() = State::Disconnected {
event: (*data).create_and_schedule_reconnect(),
reconnect_timeout: MIN_TIMEOUT_MS,
};
}
}
} else if result == 0 {
// error, signal connection close
*(*data).state.borrow_mut() = State::Disconnected {
event: (*data).create_and_schedule_reconnect(),
reconnect_timeout: MIN_TIMEOUT_MS,
};
break;
} else if result == NGX_AGAIN as isize {
if ngx_handle_read_event(conn.read, 0) != NGX_OK as isize {
*(*data).state.borrow_mut() = State::Disconnected {
event: (*data).create_and_schedule_reconnect(),
reconnect_timeout: MIN_TIMEOUT_MS,
};
}
break;
} else {
// error, retry reconnect
*(*data).state.borrow_mut() = State::Disconnected {
event: (*data).create_and_schedule_reconnect(),
reconnect_timeout: MIN_TIMEOUT_MS,
};
break;
}
Expand All @@ -530,7 +538,10 @@ unsafe extern "C" fn on_reconnect_timeout(ev: *mut ngx_event_t) {
if !data.is_null() {
let data = &*data;
let mut state = data.state.borrow_mut();
if let State::Disconnected { .. } = &*state {
if let State::Disconnected { reconnect_timeout, .. } = &mut *state {
if *reconnect_timeout < TIMEOUT_MS {
*reconnect_timeout = (*reconnect_timeout) * 2
};
if let Some(conn) = State::try_connect(
&data.path,
&data.name,
Expand All @@ -554,17 +565,18 @@ unsafe extern "C" fn on_reconnect_timeout(ev: *mut ngx_event_t) {
&& ngx_exiting == 0
&& ngx_terminate == 0
{
ngx_event_add_timer(&mut *ev, TIMEOUT_MS);
ngx_event_add_timer(&mut *ev, *reconnect_timeout);
}
State::Disconnected { event: ev }

State::Disconnected { event: ev, reconnect_timeout: *reconnect_timeout }
},
};
} else if (*ev).timer_set() == 0
&& ngx_quit == 0
&& ngx_exiting == 0
&& ngx_terminate == 0
{
ngx_event_add_timer(ev, TIMEOUT_MS);
ngx_event_add_timer(ev, *reconnect_timeout);
}
}
}
Expand Down