utilities/pueue
utilities/pueue
1
0
Fork 0
mirror of https://github.com/nukesor/pueue synced 2024-09-16 06:21:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use usize as task_id

Browse source
This commit is contained in:
Arne Beer 2019-12-16 04:50:09 +01:00
parent 9a335bc94e
commit 3de129ff69
8 changed files with 64 additions and 64 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
client/cli.rs
View file

@ -22,25 +22,25 @@ pub enum SubCommand {
/// You cannot remove running or paused tasks.
Remove {
/// The task ids to be removed
task_ids: Vec<i32>,
task_ids: Vec<usize>,
},
/// Switches the queue position two commands. Only works on queued or stashed commands
Switch {
/// The first task id
task_id_1: i32,
task_id_1: usize,
/// The second task id
task_id_2: i32,
task_id_2: usize,
},
/// Stash some tasks. These tasks won't be automatically started.
/// Afterwards either `enqueue` them, to be normally handled or forcefully `start` them.
Stash {
/// The id(s) of the tasks you want to stash
task_ids: Vec<i32>,
task_ids: Vec<usize>,
},
/// Enqueue stashed tasks. They'll be handled normally afterwards.
Enqueue {
/// The id(s) of the tasks you want to enqueue
task_ids: Vec<i32>,
task_ids: Vec<usize>,
},
/// Wake the daemon from its paused state, including continuing all paused tasks.
@ -50,12 +50,12 @@ pub enum SubCommand {
/// Doesn't affect the daemon or any other tasks.
/// Works on a paused deamon.
#[structopt(short, long)]
task_ids: Option<Vec<i32>>,
task_ids: Option<Vec<usize>>,
},
Restart {
/// Restart the
#[structopt()]
task_ids: Vec<i32>,
task_ids: Vec<usize>,
/// Start the task(s) immediately
#[structopt(name = "immediate", short, long)]
@ -72,7 +72,7 @@ pub enum SubCommand {
/// Enforce starting these tasks.
/// Doesn't affect the daemon or any other tasks.
#[structopt(short, long, group("pause"))]
task_ids: Option<Vec<i32>>,
task_ids: Option<Vec<usize>>,
},
/// Pause the daemon and all running tasks.
/// A paused daemon won't start any new tasks.
@ -85,13 +85,13 @@ pub enum SubCommand {
/// Enforce starting these tasks.
/// Doesn't affect the daemon or any other tasks.
#[structopt(group("kill"), required_unless("all"))]
task_ids: Vec<i32>,
task_ids: Vec<usize>,
},
/// Send something to a task. For example, useful for sending confirmations ('y\n')
Send {
/// The id of the task
task_id: i32,
task_id: usize,
/// The input that should be sent to the process
input: String,
@ -99,7 +99,7 @@ pub enum SubCommand {
/// Edit the command of a stashed or queued task.
Edit {
/// The id of the task
task_id: i32,
task_id: usize,
},
/// Display the current status of all tasks
@ -114,7 +114,7 @@ pub enum SubCommand {
Log {
/// Specify for which specific tasks you want to see the output
#[structopt(short, long)]
task_ids: Option<Vec<i32>>,
task_ids: Option<Vec<usize>>,
/// Print the current state as json
/// Includes EVERYTHING
#[structopt(short, long)]
@ -124,7 +124,7 @@ pub enum SubCommand {
/// This command allows following (like `tail -f`)
Show {
/// The id of the task
task_id: i32,
task_id: usize,
/// Continuously print stdout (like `tail -f`)
#[structopt(short, long)]
follow: bool,

4
client/output.rs
View file

@ -124,7 +124,7 @@ pub fn print_state(message: Message, json: bool) {
/// Print the log ouput of finished tasks.
/// Either print the logs of every task
/// or only print the logs of the specified tasks.
pub fn print_logs(message: Message, task_ids: Option<Vec<i32>>, json: bool) {
pub fn print_logs(message: Message, task_ids: Option<Vec<usize>>, json: bool) {
let state = match message {
Message::StatusResponse(state) => state,
_ => return,
@ -149,7 +149,7 @@ pub fn print_logs(message: Message, task_ids: Option<Vec<i32>>, json: bool) {
}
/// Print the log of a single task.
pub fn print_log(task_id: i32, state: &State) {
pub fn print_log(task_id: usize, state: &State) {
let task = match state.tasks.get(&task_id) {
Some(task) => task,
None => return,

8
daemon/socket/instructions.rs
View file

@ -36,7 +36,7 @@ pub fn handle_message(message: Message, sender: &Sender<Message>, state: &Shared
/// If the start_immediately flag is set, send a StartMessage to the task handler
fn add_task(message: AddMessage, sender: &Sender<Message>, state: &SharedState) -> Message {
let task = Task::new(message.command, message.path);
let task_id: i32;
let task_id: usize;
{
let mut state = state.lock().unwrap();
task_id = state.add_task(task);
@ -336,7 +336,7 @@ fn set_parallel_tasks(amount: usize, sender: &Sender<Message>) -> Message {
fn task_response_helper(
message: &'static str,
task_ids: Vec<i32>,
task_ids: Vec<usize>,
statuses: Vec<TaskStatus>,
state: &SharedState,
) -> String {
@ -354,8 +354,8 @@ fn task_response_helper(
/// and possibly tasks for which the instruction cannot be executed.
fn compile_task_response(
message: &'static str,
matching: Vec<i32>,
mismatching: Vec<i32>,
matching: Vec<usize>,
mismatching: Vec<usize>,
) -> String {
let matching: Vec<String> = matching.iter().map(|id| id.to_string()).collect();
let mismatching: Vec<String> = mismatching.iter().map(|id| id.to_string()).collect();

24
daemon/task_handler.rs
View file

@ -22,7 +22,7 @@ pub struct TaskHandler {
state: SharedState,
settings: Settings,
receiver: Receiver<Message>,
pub children: BTreeMap<i32, Child>,
pub children: BTreeMap<usize, Child>,
running: bool,
reset: bool,
}
@ -48,7 +48,7 @@ impl TaskHandler {
/// This is needed to prevent detached processes
impl Drop for TaskHandler {
fn drop(&mut self) {
let ids: Vec<i32> = self.children.keys().cloned().collect();
let ids: Vec<usize> = self.children.keys().cloned().collect();
for id in ids {
let mut child = self.children.remove(&id).expect("Failed killing children");
info!("Killing child {}", id);
@ -90,7 +90,7 @@ impl TaskHandler {
/// Return the next task that's queued for execution.
/// None if no new task could be found.
fn get_next(&mut self) -> Result<Option<(i32, Task)>> {
fn get_next(&mut self) -> Result<Option<(usize, Task)>> {
let mut state = self.state.lock().unwrap();
match state.get_next_task() {
Some(id) => {
@ -103,7 +103,7 @@ impl TaskHandler {
/// Actually spawn a new sub process
/// The output of subprocesses is piped into a seperate file for easier access
fn start_process(&mut self, task_id: i32, task: &Task) {
fn start_process(&mut self, task_id: usize, task: &Task) {
// Already get the mutex here to ensure that the task won't be manipulated
// or removed while we are starting it over here.
let mut state = self.state.lock().unwrap();
@ -202,7 +202,7 @@ impl TaskHandler {
/// Gather all finished tasks and sort them by finished and errored.
/// Returns two lists of task ids, namely finished_task_ids and errored _task_ids
fn get_finished(&mut self) -> (Vec<i32>, Vec<i32>) {
fn get_finished(&mut self) -> (Vec<usize>, Vec<usize>) {
let mut finished = Vec::new();
let mut errored = Vec::new();
for (id, child) in self.children.iter_mut() {
@ -250,7 +250,7 @@ impl TaskHandler {
}
/// Send a signal to a unix process
fn send_signal(&mut self, id: i32, signal: Signal) -> Result<bool, nix::Error> {
fn send_signal(&mut self, id: usize, signal: Signal) -> Result<bool, nix::Error> {
if let Some(child) = self.children.get(&id) {
debug!("Sending signal {} to {}", signal, id);
let pid = Pid::from_raw(child.id() as i32);
@ -291,7 +291,7 @@ impl TaskHandler {
}
// Start the daemon and all paused tasks
let keys: Vec<i32> = self.children.keys().cloned().collect();
let keys: Vec<usize> = self.children.keys().cloned().collect();
for id in keys {
self.continue_task(id);
}
@ -305,7 +305,7 @@ impl TaskHandler {
}
/// Send a start signal to a paused task to continue execution
fn continue_task(&mut self, id: i32) {
fn continue_task(&mut self, id: usize) {
if !self.children.contains_key(&id) {
return;
}
@ -333,7 +333,7 @@ impl TaskHandler {
}
// Pause the daemon and all tasks
let keys: Vec<i32> = self.children.keys().cloned().collect();
let keys: Vec<usize> = self.children.keys().cloned().collect();
if !message.wait {
for id in keys {
self.pause_task(id);
@ -350,7 +350,7 @@ impl TaskHandler {
/// Pause a specific task.
/// Send a signal to the process to actually pause the OS process
fn pause_task(&mut self, id: i32) {
fn pause_task(&mut self, id: usize) {
if !self.children.contains_key(&id) {
return;
}
@ -379,7 +379,7 @@ impl TaskHandler {
// Pause the daemon and all tasks
info!("Killing all spawned children");
let keys: Vec<i32> = self.children.keys().cloned().collect();
let keys: Vec<usize> = self.children.keys().cloned().collect();
for id in keys {
self.kill_task(id);
}
@ -387,7 +387,7 @@ impl TaskHandler {
/// Kill a specific task and handle it accordingly
/// Triggered on `reset` and `kill`.
fn kill_task(&mut self, task_id: i32) {
fn kill_task(&mut self, task_id: usize) {
if let Some(child) = self.children.get_mut(&task_id) {
match child.kill() {
Err(_) => debug!("Task {} has already finished by itself", task_id),

10
shared/log.rs
View file

@ -6,14 +6,14 @@ use ::std::path::{Path, PathBuf};
use crate::settings::Settings;
pub fn get_log_paths(task_id: i32, settings: &Settings) -> (PathBuf, PathBuf) {
pub fn get_log_paths(task_id: usize, settings: &Settings) -> (PathBuf, PathBuf) {
let pueue_dir = Path::new(&settings.daemon.pueue_directory).join("temp");
let out_path = pueue_dir.join(format!("{}_stdout.log", task_id));
let err_path = pueue_dir.join(format!("{}_stderr.log", task_id));
(out_path, err_path)
}
pub fn create_log_file_handles(task_id: i32, settings: &Settings) -> Result<(File, File)> {
pub fn create_log_file_handles(task_id: usize, settings: &Settings) -> Result<(File, File)> {
let (out_path, err_path) = get_log_paths(task_id, settings);
let stdout = File::create(out_path)?;
let stderr = File::create(err_path)?;
@ -21,7 +21,7 @@ pub fn create_log_file_handles(task_id: i32, settings: &Settings) -> Result<(Fil
Ok((stdout, stderr))
}
pub fn get_log_file_handles(task_id: i32, settings: &Settings) -> Result<(File, File)> {
pub fn get_log_file_handles(task_id: usize, settings: &Settings) -> Result<(File, File)> {
let (out_path, err_path) = get_log_paths(task_id, settings);
let stdout = File::open(out_path)?;
let stderr = File::open(err_path)?;
@ -29,7 +29,7 @@ pub fn get_log_file_handles(task_id: i32, settings: &Settings) -> Result<(File,
Ok((stdout, stderr))
}
pub fn read_log_files(task_id: i32, settings: &Settings) -> Result<(String, String)> {
pub fn read_log_files(task_id: usize, settings: &Settings) -> Result<(String, String)> {
let (mut stdout_handle, mut stderr_handle) = get_log_file_handles(task_id, settings)?;
let mut stdout = String::new();
let mut stderr = String::new();
@ -40,7 +40,7 @@ pub fn read_log_files(task_id: i32, settings: &Settings) -> Result<(String, Stri
Ok((stdout, stderr))
}
pub fn clean_log_handles(task_id: i32, settings: &Settings) {
pub fn clean_log_handles(task_id: usize, settings: &Settings) {
let (out_path, err_path) = get_log_paths(task_id, settings);
if let Err(err) = remove_file(out_path) {
error!(

28
shared/message.rs
View file

@ -45,74 +45,74 @@ pub struct AddMessage {
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct RemoveMessage {
pub task_ids: Vec<i32>,
pub task_ids: Vec<usize>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct SwitchMessage {
pub task_id_1: i32,
pub task_id_2: i32,
pub task_id_1: usize,
pub task_id_2: usize,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct StashMessage {
pub task_ids: Vec<i32>,
pub task_ids: Vec<usize>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct EnqueueMessage {
pub task_ids: Vec<i32>,
pub task_ids: Vec<usize>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct StartMessage {
pub task_ids: Option<Vec<i32>>,
pub task_ids: Option<Vec<usize>>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct RestartMessage {
pub task_ids: Vec<i32>,
pub task_ids: Vec<usize>,
pub start_immediately: bool,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct PauseMessage {
pub wait: bool,
pub task_ids: Option<Vec<i32>>,
pub task_ids: Option<Vec<usize>>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct KillMessage {
pub all: bool,
pub task_ids: Vec<i32>,
pub task_ids: Vec<usize>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct SendMessage {
pub task_id: i32,
pub task_id: usize,
pub input: String,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct EditMessage {
pub task_id: i32,
pub task_id: usize,
pub command: String,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct EditRequestMessage {
pub task_id: i32,
pub task_id: usize,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct EditResponseMessage {
pub task_id: i32,
pub task_id: usize,
pub command: String,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct StreamRequestMessage {
pub task_id: i32,
pub task_id: usize,
pub follow: bool,
pub err: bool,
}

26
shared/state.rs
View file

@ -16,10 +16,10 @@ pub type SharedState = Arc<Mutex<State>>;
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct State {
max_id: i32,
max_id: usize,
settings: Settings,
pub running: bool,
pub tasks: BTreeMap<i32, Task>,
pub tasks: BTreeMap<usize, Task>,
}
impl State {
@ -34,7 +34,7 @@ impl State {
state
}
pub fn add_task(&mut self, mut task: Task) -> i32 {
pub fn add_task(&mut self, mut task: Task) -> usize {
task.id = self.max_id;
self.tasks.insert(self.max_id, task);
self.max_id += 1;
@ -42,11 +42,11 @@ impl State {
self.max_id - 1
}
pub fn remove_task(&mut self, id: i32) -> Option<Task> {
pub fn remove_task(&mut self, id: usize) -> Option<Task> {
self.tasks.remove(&id)
}
pub fn get_task_clone(&mut self, id: i32) -> Option<Task> {
pub fn get_task_clone(&mut self, id: usize) -> Option<Task> {
let task = self.tasks.remove(&id);
let clone = task.clone();
if let Some(task) = task {
@ -56,7 +56,7 @@ impl State {
return clone;
}
pub fn get_next_task(&mut self) -> Option<i32> {
pub fn get_next_task(&mut self) -> Option<usize> {
for (id, task) in self.tasks.iter() {
if task.status == TaskStatus::Queued {
return Some(*id);
@ -65,7 +65,7 @@ impl State {
None
}
pub fn change_status(&mut self, id: i32, new_status: TaskStatus) {
pub fn change_status(&mut self, id: usize, new_status: TaskStatus) {
if let Some(ref mut task) = self.tasks.get_mut(&id) {
if new_status == TaskStatus::Running {
if TaskStatus::Queued == task.status || TaskStatus::Stashed == task.status {
@ -77,13 +77,13 @@ impl State {
};
}
pub fn add_error_message(&mut self, id: i32, message: String) {
pub fn add_error_message(&mut self, id: usize, message: String) {
if let Some(ref mut task) = self.tasks.get_mut(&id) {
task.stderr = Some(message);
}
}
pub fn get_task_status(&mut self, id: i32) -> Option<TaskStatus> {
pub fn get_task_status(&mut self, id: usize) -> Option<TaskStatus> {
if let Some(ref task) = self.tasks.get(&id) {
return Some(task.status.clone());
};
@ -97,9 +97,9 @@ impl State {
/// Additionally, if no task_ids are specified, return ids of all tasks
pub fn tasks_in_statuses(
&mut self,
task_ids: Option<Vec<i32>>,
task_ids: Option<Vec<usize>>,
statuses: Vec<TaskStatus>,
) -> (Vec<i32>, Vec<i32>) {
) -> (Vec<usize>, Vec<usize>) {
let task_ids = match task_ids {
Some(ids) => ids,
None => self.tasks.keys().cloned().collect(),
@ -132,9 +132,9 @@ impl State {
/// The same as tasks_in_statuses, but with inverted statuses
pub fn tasks_not_in_statuses(
&mut self,
task_ids: Option<Vec<i32>>,
task_ids: Option<Vec<usize>>,
excluded_statuses: Vec<TaskStatus>,
) -> (Vec<i32>, Vec<i32>) {
) -> (Vec<usize>, Vec<usize>) {
let mut valid_statuses = Vec::new();
// Create a list of all valid statuses
// (statuses that aren't the exl

2
shared/task.rs
View file

@ -21,7 +21,7 @@ pub enum TaskStatus {
/// Upon task completion, the output is read from the files and put into the struct.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Task {
pub id: i32,
pub id: usize,
pub command: String,
pub path: String,
pub status: TaskStatus,