Hello,

I need to write an Elixir function which will evenly distribute unbalanced load over set of tasks (nodes). Lets say I have following input:

number_nodes = 3
weighted_tasks = [10, 10, 10, 7, 6, 5, 5, 2, 2, 1, 1, 1]
node_capacity = 20 # as sum of all tasks/number_nodes

and I need to get this output (one of other possible combinations)

[[10, 10], [10, 7, 2, 1], [6, 5, 5, 2, 1, 1]]

which mean that load weight per node is 20 for all 3 nodes and balanced.

I am trying Enum.chunck_by() function but not able implement it correctly, here is my attempt:

  def chuck_f(node_capacity) do
    chunk_fun = fn element, acc ->
      acc_capacity = Enum.sum(acc)
      pending_capacity = acc_capacity + element

      if pending_capacity < node_capacity do
        # do not yet emit given chuck, but remove this element from the input list as it
        # is added to the chuck, but how to do that?
        {:cont, [element | acc]}
      else
        # if no more element in input list present emit chunck
        # if there are some element just skip current one and do not emit chunck
        # but how to get it?
        {:cont, [acc], []}
      end
    end
    chunk_fun
  end


  def after_iteration_fun() do
    after_fun = fn acc ->
        case acc do
          [] -> {:cont, []}
          acc -> {:cont, acc, []}
        end
      end
    after_fun
  end

  def distribute_load do
    input = [10, 10, 10, 7, 6, 5, 5, 2, 2, 1, 1, 1]
    node_capacity = 20
    input |> Enum.chunk_while([], chuck_f(node_capacity), after_iteration_fun())
  end

which producing result:

iex(109)> distribute_load
[[[10]], [[7, 10]], [1, 1, 1, 2, 2, 5, 5]]

which is not what I need.

I have this algorithm implement in Java with this pseudo code:

partitions_list = compute_list_of_partiton_with_load_sort_desc # this is already done in Elixir
cluster_load = compute_total_cluster_load
one_node_load = cluster_load/num_of_nodes


#for all nodes in cluster
for node in node_list {
    
    # initiate node with no partitions yet and full capacity
    node.partion_list = []
    node.capacity = one_node_load

    for partion in partition_list {

        if partion.load <= node.capacity {	
            node.partion_list.add(partition)
            recompute_node_capacity(node, partition)
            partition_list.remove(partition)

        } 
    }
}


def recompute_node_capacity(node, partitions) do
    new_capacit = node.capacity - partition.load
    
    node.capacity = new_capacity # ugly node's state mutation
end

and this actual draft Java impl where I can easily mutate map/list just in place which give me ability to achieve the goal:

@Test
    public void balanceLoad() {
        
        List<Integer> tasks = new ArrayList<>(); 
        tasks.add(10);
        tasks.add(10);
        tasks.add(10);
        tasks.add(7);
        tasks.add(6);
        tasks.add(5);
        tasks.add(5);
        tasks.add(2);
        tasks.add(2);
        tasks.add(1);
        tasks.add(1);
        tasks.add(1);        
        
        Map<Integer, List<Integer>> nodes = new HashMap<>();
        nodes.put(1, new ArrayList<>());
        nodes.put(2, new ArrayList<>());
        nodes.put(3, new ArrayList<>());
        
        
        System.out.println("Before: " + nodes);
                
        for (int nodeKey : nodes.keySet()) {
            List<Integer> nodeList = new ArrayList<>(tasks);
            List<Integer> nodeTasks = nodes.get(nodeKey);
            
            for (Iterator<Integer> iterator = nodeList.iterator(); iterator.hasNext();) {
                Integer task = iterator.next();
         
                if(task <= nodeCapacity(nodeTasks, 20)) {
                    nodeTasks.add(task);
                    iterator.remove();
                    tasks.remove(task);
                }
            }
        }
        System.out.println("After: " + nodes);
    }
    
    
    private int nodeCapacity(List<Integer> nodeTasks, int nodeCapCapacity) {
        int current = nodeTasks.stream().reduce(0, Integer::sum);
        return nodeCapCapacity - current;
    }

giving me this result:

Before: {1=, 2=, 3=}
After: {1=[10, 10], 2=[10, 7, 2, 1], 3=[6, 5, 5, 2, 1, 1]}

Can anybody assist to achieve similar in Elixir? Thank you for your time.