Dataloaders with socket conection

[dfalbel]

Passing data between the workers can now be accelerated by using socket connections.

> bench_ds <- dataset(
+   initialize = function(size_of_tensor) {
+     self$size_of_tensor <- size_of_tensor
+   },
+   .getitem = function(i) {
+     list(x = torch_randn(self$size_of_tensor))
+   },
+   .length = function() {32*20}
+ )
> system.time({
+   withr::with_options(list(torch.dataloader_use_socket_con = TRUE), {
+     dl <- dataloader(bench_ds(256*256*3), batch_size = 32, num_workers = 4)
+     coro::loop(for(x in dl) {
+       k <- x
+     })
+   })
+ })
   user  system elapsed 
 48.416   2.419  58.796 
> system.time({
+   withr::with_options(list(torch.dataloader_use_socket_con = FALSE), {
+     dl <- dataloader(bench_ds(256*256*3), batch_size = 32, num_workers = 4)
+     coro::loop(for(x in dl) {
+       k <- x
+     })
+   })
+ })
   user  system elapsed 
 61.417   2.273  74.165 

Next steps should be investing in torch_save speedups, specially, if we could avoid the compression.
Perhaps, also finding a different way of serializing the tensor.

[dfalbel]

With further improvements in the serialization function we get:

> system.time({
+   withr::with_options(list(torch.dataloader_use_socket_con = TRUE), {
+     dl <- dataloader(bench_ds(256*256*3), batch_size = 32, num_workers = 4)
+     coro::loop(for(x in dl) {
+       k <- x
+     })
+   })
+ })
   user  system elapsed 
  2.330   0.942   6.623 
> system.time({
+   withr::with_options(list(torch.dataloader_use_socket_con = FALSE), {
+     dl <- dataloader(bench_ds(256*256*3), batch_size = 32, num_workers = 4)
+     coro::loop(for(x in dl) {
+       k <- x
+     })
+   })
+ })
   user  system elapsed 
  4.409   0.814  14.548

[dfalbel]

Further improvements from callr (r-lib/callr#223) make it slightly faster, specially when not using the socket conections.

> system.time({
+   withr::with_options(list(torch.dataloader_use_socket_con = TRUE), {
+     dl <- dataloader(bench_ds(256*256*3), batch_size = 32, num_workers = 4)
+     coro::loop(for(x in dl) {
+       k <- x
+     })
+   })
+ })
   user  system elapsed 
  2.248   0.987   6.796 
> system.time({
+   withr::with_options(list(torch.dataloader_use_socket_con = FALSE), {
+     dl <- dataloader(bench_ds(256*256*3), batch_size = 32, num_workers = 4)
+     coro::loop(for(x in dl) {
+       k <- x
+     })
+   })
+ })
   user  system elapsed 
  2.166   0.896   8.003