Merge pull request #835 from kitsudaiki/feat/improve-task-processing

related issue: #834

Author: kitsudaiki

CHANGELOG.md

@@ -27,6 +27,7 @@
     - quota-management was added to the admin-section of the dashboard
     - dataset list in the dashboard now prints the number of rows and columns
     - cluster list in the dashboard now prints the cluster adress
+- added forecast-length as new parameter for train-tasks (only via python-sdk at the moment)
 
 ### Changed
 
@@ -46,7 +47,8 @@
 - fixed broken version-output of the pre-build docker images
 - restart of hanami and sakura instance in regard of the cluster was now fixed to avoid broken cluster and hosts after restart
 - restarts of the same sakura-host doesn't result in duplications in hanami anymore
-
+- progress-bar in dashboard for multiple epochs is now calculated correctly
+- fixed number of output-values in case of int and float output-type
 
 ## v0.10.0
 

Cargo.toml

@@ -24,5 +24,5 @@ overflow-checks = true
 [profile.release]
 opt-level = 3
 lto = "fat"
-codegen-units = 1
+codegen-units = 8
 strip = true

src/binaries/sakura/src/api/http_endpoints/model/task/checkpoint_restore_task_v1_0.rs

@@ -88,7 +88,7 @@ pub async fn checkpoint_restore_task(
         model_uuid: *model_uuid,
         name: body.name.clone(),
         info: TaskVariant::CheckpointRestore(info),
-        meta: TaskMeta::new(1, 1, 1),
+        meta: TaskMeta::new(1, 1, 1, 0),
     };
     super::add_task_to_model(task, &task_type, &context)?;
 

src/binaries/sakura/src/api/http_endpoints/model/task/checkpoint_save_task_v1_0.rs

@@ -89,7 +89,7 @@ pub async fn checkpoint_save_task(
         model_uuid: *model_uuid,
         name: body.name.clone(),
         info: TaskVariant::CheckpointSave(info),
-        meta: TaskMeta::new(1, 1, 1),
+        meta: TaskMeta::new(1, 1, 1, 0),
     };
     super::add_task_to_model(task, &task_type, &context)?;
 

src/binaries/sakura/src/api/http_endpoints/model/task/create_request_task_v1_0.rs

@@ -141,7 +141,7 @@ pub async fn create_request_task(
             model_uuid: *model_uuid,
             name: body.name.clone(),
             info: TaskVariant::Request(Box::new(info)),
-            meta: TaskMeta::new(number_of_cycles, 1, time_length),
+            meta: TaskMeta::new(number_of_cycles, 1, time_length, 0),
         };
         super::add_task_to_model(task, &task_type, &context)?;
 

src/binaries/sakura/src/api/http_endpoints/model/task/create_train_task_v1_0.rs

@@ -53,6 +53,7 @@ pub async fn create_train_task(
     let task_uuid = Uuid::new_v4();
     let task_type = TaskType::Train;
     let time_length = body.time_length.unwrap_or(1);
+    let forecast_length = body.forecast_length.unwrap_or(0);
     let mut number_of_cycles = u64::MAX;
 
     if time_length < 1 {
@@ -111,21 +112,31 @@ pub async fn create_train_task(
         }
 
         // handle the time-lenght-value
-        if number_of_cycles < time_length {
+        let cycle_length = time_length + forecast_length;
+        if number_of_cycles < cycle_length {
             let msg = format!(
-                "Time-length {time_length} is bigger than at least of of the seleced datasets."
+                "Time-length {cycle_length} is bigger than at least of of the seleced datasets."
             );
             return Err(ErrorResponse::BadRequest(msg));
         }
+        // TODO: seems not fully correct calculated
         number_of_cycles -= time_length - 1;
+        if forecast_length > 0 {
+            number_of_cycles /= forecast_length;
+        }
 
         // create new task
         let task = Task {
             uuid: task_uuid,
             model_uuid: *model_uuid,
             name: body.name.clone(),
             info: TaskVariant::Training(info),
-            meta: TaskMeta::new(number_of_cycles, body.number_of_epochs, time_length),
+            meta: TaskMeta::new(
+                number_of_cycles,
+                body.number_of_epochs,
+                time_length,
+                forecast_length,
+            ),
         };
         super::add_task_to_model(task, &task_type, &context)?;
 

src/binaries/sakura/src/core/blocks/core_block.rs

@@ -42,9 +42,8 @@ use super::super::processing::worker_queue::*;
 pub struct Synapse {
     /// The threshold value that must be exceeded for the synapse to activate.
     pub border: f32,
-    /// First synaptic weight value.
+    /// synaptic weight values.
     pub weight_1: f32,
-    /// Second synaptic weight value.
     pub weight_2: f32,
 
     /// Counter tracking how many times this synapse has been activated.

src/binaries/sakura/src/core/blocks/input_block.rs

@@ -144,14 +144,14 @@ impl InputBlock {
     /// * `input_size` - Size of the input data.
     /// * `offset` - Offset to apply to the input data.
     /// * `time_length` - Length of the time dimension of the input data.
-    /// * `allow_cration` - Whether to allow creation of new input links.
+    /// * `allow_creation` - Whether to allow creation of new input links.
     pub fn apply_input(
         &mut self,
         input_ptr: &[f32],
         input_size: usize,
         offset: usize,
         time_length: usize,
-        allow_cration: bool,
+        allow_creation: bool,
     ) {
         // resize links, if necessary
         let maximum_size = input_size * 2 * time_length;
@@ -162,7 +162,7 @@ impl InputBlock {
         let mut is_negative;
         let mut total_position;
 
-        if allow_cration {
+        if allow_creation {
             // update links
             for (i, val) in input_ptr.iter().enumerate().take(input_size) {
                 total_position = (offset + i) * 2;

src/binaries/sakura/src/core/model_interface.rs

@@ -218,7 +218,6 @@ impl ModelInterface {
                 model_data_handler.get_output_buffer(&self.model_uuid, hexagon_name)?;
 
             let mut output_buffer = output_buffer_mutex.lock().expect("mutex poisoned");
-            data.resize(output_buffer.output_neurons.len(), 0.0f32);
             convert_output_to_buffer(data, &mut output_buffer);
         }
 

src/binaries/sakura/src/core/processing/finish_counter.rs

@@ -21,25 +21,12 @@ use super::tasks::Task;
 /// within a specific cycle, and can trigger actions when the completion criteria are met.
 #[derive(Default, Debug)]
 pub struct FinishCounter {
-    /// Number of input tasks that have been compared in the current cycle.
     pub input_compare: usize,
-
-    /// Number of output tasks that have been compared in the current cycle.
     pub output_compare: usize,
-
-    /// The threshold number of task comparisons needed to consider the cycle finished.
     task_compare: usize,
-
-    /// Current count of completed tasks in the current cycle.
     counter: usize,
-
-    /// The cycle number we're expecting to complete.
     expected_cycle_number: u64,
-
-    /// Flag indicating whether this cycle has already been marked as finished.
     already_finished: bool,
-
-    /// Optional reference to the task that this counter is associated with.
     pub task: Option<Arc<Mutex<Task>>>,
 }