Distributed function processing with estimate-based scheduler
Systems and methods for distributed storage and processing systems using estimate-based schedulers are described. A node receives estimated processing data for each storage device including redundant copies of data chunks for a data unit. The node determines, based on the estimated processing data and data paths to each data chunk, a task time estimate for data paths to each data chunk and selects data paths for at least one copy of each data chunk to be processed using a corresponding set of compute tasks. The compute tasks are sent for processing based on the assignments of the node.
1. A system, comprising:
at least one memory;
at least one processor; and
a host controller, stored in the at least one memory and executable by the at least one processor, to perform operations including:
receiving estimated processing data for a plurality of storage nodes, wherein:
the plurality of storage nodes is configured to include redundant copies of a plurality of target data chunks for processing in a set of compute tasks;
the set of compute tasks is configured to target a set of target data chunks from the redundant copies of the plurality of target data chunks;
the set of target data chunks is configured to include at least one copy from the redundant copies for each target data chunk in the plurality of target data chunks; and
the estimated processing data is configured to include a soft error estimate of correctable errors in a data retrieval process for each storage node of the plurality of storage nodes;
determining, based on estimated processing data and a plurality of data paths to redundant copies of each target data chunk, a task time estimate for each data path of the plurality of data paths;
selecting at least one data path for each target data chunk in the set of target data chunks for processing in the set of compute tasks; and
sending the set of compute tasks using the selected at least one data path for each target data chunk in the set of target data chunks.
2. The system of claim 1 , wherein:
the operations further include:
sending, during a processing period for a prior set of compute tasks, an estimate request to each storage node of the plurality of storage nodes;
receiving the estimated processing data for the plurality of storage nodes includes receiving, during the processing period for the prior set of compute tasks, the estimated processing data for at least one compute task of the set of compute tasks from each storage node of the plurality of storage nodes; and
the set of compute tasks follows the prior set of compute tasks in a series of compute tasks processed by the host controller.
3. The system of claim 1 , wherein the estimated processing data is further configured to include a plurality of estimate values selected from:
a task processing time estimate;
a latency estimate; and
a hard error estimate.
4. The system of claim 1 , wherein:
the estimated processing data is further configured to include a task processing time estimate and a latency estimate;
determining the task time estimate for each data path of the plurality of data paths is further based on the task processing time estimate, the latency estimate, and the soft error estimate; and
the operations further include:
comparing, for redundant copies of a target data chunk, the task time estimate for each data path to select the at least one data path to the target data chunk.
5. The system of claim 1 , wherein the operations further include:
determining a processing time threshold for the set of compute tasks: and
selecting, using the processing time threshold, the at least one data path for each target data chunk.
6. The system of claim 5 , wherein:
each data path to each storage node of the plurality of storage nodes is configured to include a data processing node configured to execute a compute task for a corresponding data chunk in a target storage device; and
the operations further include:
determining a combined task time for at least two compute tasks for a target data processing node;
comparing the combined task time to the processing time threshold for the set of compute tasks; and
selecting, responsive to the processing time threshold meeting at least the combined task time, data paths through the target data processing node for target data chunks for the at least two compute tasks.
7. The system of claim 5 , wherein:
each target data chunk has a corresponding compute task in the set of compute tasks; and
the operations further include, responsive to the task time estimate for at least one data path to redundant copies of a target data chunk exceeding the processing time threshold:
splitting the target data chunk into at least two data chunk portions;
splitting a corresponding compute task into at least two compute task portions;
selecting a first data path for a first storage node from the plurality of storage nodes for a first data chunk portion of the at least two data chunk portions; and
selecting a second data path for a second storage node from the plurality of storage nodes for a second data chunk portion of the at least two data chunk portions.
8. The system of claim 5 , wherein:
each target data chunk has a corresponding compute task in the set of compute tasks; and
the operations further include, responsive to the task time estimate for at least one data path to redundant copies of a target data chunk exceeding the processing time threshold:
selecting a first data path for a first storage node from the plurality of storage nodes for the target data chunk and corresponding compute task;
selecting a second data path for a second storage node from the plurality of storage nodes for the target data chunk and corresponding compute task; and
returning a compute result for a first completed corresponding compute task from between the first data path and the second data path.
9. The system of claim 1 , wherein the operations further include:
receiving actual task time values from the set of compute tasks that used the selected at least one data path for each target data chunk;
determining at least one variance between actual task time values and task time estimates;
determining, based on the at least one variance, at least one correction factor for an estimator used to generate task time estimates; and
sending the at least one correction factor to the estimator for generating future task time estimates.
10. The system of claim 1 , further comprising:
the plurality of storage nodes configured to:
store, in a plurality of storage devices, a target data file, wherein the redundant copies of data chunks correspond to redundant portions of the target data file; and
determine estimated processing data for the set of compute tasks; and
a plurality of data processing nodes configured to:
process the set of compute tasks using the target data chunks, wherein the at least one data path to each target data chunk includes at least one corresponding data processing node of the plurality of data processing nodes; and
return a compute task result for each compute task of the set of compute tasks.
11. A computer-based method, comprising:
receiving estimated processing data for a plurality of storage nodes, wherein:
the plurality of storage nodes includes redundant copies of a plurality of target data chunks for processing in a set of compute tasks;
the set of compute tasks targets a set of target data chunks from the redundant copies of the plurality of target data chunks;
the set of target data chunks includes at least one copy from the redundant copies for each target data chunk in the plurality of target data chunks; and
the estimated processing data includes a soft error estimate of correctable errors in a data retrieval process for each storage node of the plurality of storage nodes;
determining, based on estimated processing data and a plurality of data paths to redundant copies of each target data chunk, a task time estimate for each data path of the plurality of data paths;
selecting at least one data path for each target data chunk in the set of target data chunks to be processed using the set of compute tasks; and
sending the set of compute tasks for processing using the selected at least one data path for each target data chunk in the set of target data chunks.
12. The computer-based method of claim 11 , further comprising:
sending, during a processing period for a prior set of compute tasks, an estimate request to each storage device of the plurality of storage nodes, wherein:
receiving the estimated processing data for the plurality of storage nodes includes receiving, during the processing period for the prior set of compute tasks, the estimated processing data for at least one compute task of the set of compute tasks from each storage node of the plurality of storage nodes; and
the set of compute tasks follows the prior set of compute tasks in a series of compute tasks.
13. The computer-based method of claim 11 , wherein the estimated processing data further includes a plurality of estimate values selected from:
a task processing time estimate;
a latency estimate; and
a hard error estimate.
14. The computer-based method of claim 11 , further comprising:
comparing, for redundant copies of a target data chunk, the task time estimate for each data path to select the at least one data path to the target data chunk, wherein:
the estimated processing data further includes a task processing time estimate and a latency estimate; and
determining the task time estimate for each data path of the plurality of data paths is further based on the task processing time estimate and the latency estimate.
15. The computer-based method of claim 11 , further comprising:
determining a processing time threshold for the set of compute tasks: and
selecting, using the processing time threshold, the at least one data path for each target data chunk.
16. The computer-based method of claim 15 , further comprising:
determining a combined task time for at least two compute tasks for a data processing node;
comparing the combined task time to the processing time threshold for the set of compute tasks; and
selecting, responsive to the processing time threshold meeting at least the combined task time, data paths through the data processing node for target data chunks for the at least two compute tasks.
17. The computer-based method of claim 15 , further comprising, responsive to the task time estimate for at least one data path to redundant copies of a target data chunk exceeding the processing time threshold:
splitting the target data chunk into at least two data chunk portions;
splitting a corresponding compute task into at least two compute task portions;
selecting a first data path for a first storage device from the plurality of storage nodes for a first data chunk portion of the at least two data chunk portions; and
selecting a second data path for a second storage device from the plurality of storage nodes for a second data chunk portion of the at least two data chunk portions.
18. The computer-based method of claim 15 , further comprising, responsive to the task time estimate for at least one data path to redundant copies of a target data chunk exceeding the processing time threshold:
selecting a first data path for a first storage node from the plurality of storage nodes for the target data chunk and corresponding compute task;
selecting a second data path for a second storage node from the plurality of storage nodes for the target data chunk and corresponding compute task; and
returning a compute result for a first completed corresponding compute task from between the first data path and the second data path.
19. The computer-based method of claim 11 , further comprising:
receiving actual task time values from the set of compute tasks that used the selected at least one data path for each target data chunk;
determining at least one variance between actual task time values and task time estimates;
determining, based on the at least one variance, at least one correction factor for an estimator used to generate task time estimates; and
sending the at least one correction factor to the estimator for generating future task time estimates.
20. A system, comprising:
a plurality of storage nodes configured to store, in a plurality of storage devices, a target data file as redundant copies of a plurality of target data chunks;
a plurality of data processing nodes corresponding to the plurality of storage nodes and configured to process a set of compute tasks using the plurality of target data chunks for the target data file, wherein:
the set of compute tasks targets a set of target data chunks from the redundant copies of the plurality of target data chunks; and
the set of target data chunks includes at least one copy from the redundant copies for each target data chunk in the plurality of target data chunks for the target data file;
means for receiving estimated processing data for the plurality of storage devices for each redundant copy of the redundant copies of data chunks, wherein the estimated processing data includes a soft error estimate of correctable errors in a data retrieval process for each storage device of the plurality of storage devices;
means for determining, based on the estimated processing data and a plurality of data paths to redundant copies of each target data chunk, a task time estimate for each data path of the plurality of data paths;
means for selecting at least one data path for each data chunk in the set of target data chunks to be processed using the set of compute tasks; and
means for sending the set of compute tasks for processing using the selected at least one data path for each data chunk in the set of target data chunks.