multi machine training, Met all requirement except a shared filesystem for checkpoint directory | Any workaround

[rinisin13]

Steps to reproduce

1. ---

2. ---

3. ---

Observed Results

• What happened? This could be a description, log output, etc.

Expected Results

• What did you expect to happen?

Relevant Code

// TODO(you): code here to reproduce the problem

[adamlerer]

Technically if you're running in multi-machine mode you don't need to share any files. So if you create the same path on each machine, and put all the input files (e.g. edgelists) in the same place on all the machines, it might work, although there will probably be a couple small issues to fix in the code.

Could you try it out and report any problems you run into?

[sgatamex]

If the input data is huge then putting and copying all the data in each machine may cause a problem. So technically, multimachine mode can work in independently and does not require common checkpoint directory to write to but if you need to store intermediate embeddings for future use in downstream machine learning tasks then we need common place to store these files.

Model update
i have seen in the code that model is updated from rank 0 only so if we don't use checkpoint directory it will still work, we just need to take the model from the rank 0. Same is true for stats.

Training Resume From Last Run
How this feature will work , not sure if there is no checkpoint?

Hadoop / HDFS Support without direct access to shared checkpoint

i am working on to use PBG from spark over hadoop /hdfs without common checkpoint directory access and this should be cases with other scenarios where you may have lots of machine but not having NFS mount for various reasons.

I am thinking of to have a server which will issue lease / license with expiry period to individual machines on network. So whenever they want to read / write to a common location, these machines can request a license from the server and server will manage the licenses including when to issue the license and when to ask for waiting.

Hadoop allows to copy and paste data from individual node using hadoop fs command so after getting the license from the server, individual machine can start transfer the files to shared location using hadoop commands. This will also eliminate the need for copying all data to the individual machines.

There are some issues /difficulties with this approach but we can try, any thoughts please?

[sgatamex]

Any suggestions for above approach?

[rinisin13]

@adamlerer Thanks for suggestions. Could you please help us in understand if there is no share filesystem then how partial embedding created by some bucket B1 with partition (P1,P2) in machine M1 will be available to other machine M2 while processing lets say bucket B2 with partition (P2,P4) ? do we need to have some mechanism to copy partial embedding created among all machines as suggested by @sgatamex ?

[adamlerer]

@sgatamex if you want Hadoop to act as the "shared storage" instead of a shared filesystem, then you can write your own storage backend to handle that. We wrote a storage abstraction to handle this (we use some different storage backends at Facebook as well). If you look here,

https://github.com/facebookresearch/PyTorch-BigGraph/blob/master/torchbiggraph/graph_storages.py

you can write your own subclasses HadoopEntityStorage, HadoopRelationTypeStorage, etc. and then in the config you can just set the paths to "hadoop://my/path" and it will use hadoop. It's a bit of work to write these classes but probably worth it! And it would be amazing if you could contribute these back.

@lw wrote the graph storage abstractions, so he might have more to add.

@rinisin13 during multi-machine training, embeddings are communicated between machines using the torch.distributed package ( https://pytorch.org/docs/stable/distributed.html ), which can use multiple transport backends ( TCP, MPI, or gloo ). Communicating the embeddings back and forth via Hadoop will probably be quite slow...

[sgatamex]

@adamlerer Thank you so much for your suggestions , I will look into it. I am also thinking about putting a lease server in between PBG trainers and Hadoop/HDFS system so that accidently no mess up with the data written by e.g. HadooCheckPointStorage. I know these trainers are writing into its own files but it cud be possible that actual network may be slow and request may not handled in order.
Any thoughts?

[lw]

In a different scenario I have heard of people overcoming the lack of a shared filesystem by creating an ephemeral one between the trainer nodes: before starting training, all nodes start a NFS daemon, dedicating part of their local disk to it, which then connects to the other daemons and creates a shared volume. The trainer then can write and read to this volume. Once training completes, data from this volume is copied to a more permanent storage. Unfortunately I don't have pointers on how to do you, you're going to have to search for it yourself...

Also note that this is a worse solution than a real shared filesystem: since it is used for checkpointing, its purpose is to hold data safe in case the training stops and must be restarted. By storing checkpoints on the same machines as the trainers, you run the risk that if a trainer machine fails you will also lose the checkpoint and it will all have been in vain.

Therefore I think the best solution would still be to write your own HadoopFooStorage classes to directly write to a permanent storage: it may be more upfront work, but it should pay off in terms of reliability in the longer term.

As for your concerns about data races during storage, the checkpoint format is designed in such a way that a file is only written to once and then never modified or overwritten (the only exception to this being the tiny .txt file that stores the checkpoint version). And, also, each file is only written to by one single machine. I believe with these two assumptions it might be easier to write a robust storage implementation.

[sgatamex]

@lw, thank you for your input. I agree to write the own implementation of AbstractStorage for hadoop and we are trying that only from start.

I still feel the need for synchronizing read/write requests for common storage through an extra layer of indirection which is very fast and just let trainers know if its ok to perform the operation or to wait since there is some operation still pending on the protected resource so that our data double guaranteed protected from accidental read/write. I know PBG works in a way that most of common files (checkpoint_version, traing_stats) are only updated through rank 0 trainer so there is guaranteed no races but hadoop commands sometime returns with success but still background copying of the data is still pending/going on (these are rarest of rarest scenarios but happened in the past ).So in those cases, i really feel having an extra layer won't hurt so much on the performance but at the same time provide shield against dirty reads / writes but this approach contradicts hogwild style of training used in PBG.

[Sherler]

Technically if you're running in multi-machine mode you don't need to share any files. So if you create the same path on each machine, and put all the input files (e.g. edgelists) in the same place on all the machines, it might work, although there will probably be a couple small issues to fix in the code.

Could you try it out and report any problems you run into?

I tried it and result with the following error at rank none-0 trainer：

Traceback (most recent call last):
  File "/root/miniconda3/bin/torchbiggraph_train", line 8, in <module>
    sys.exit(main())
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/train.py", line 71, in main
    train(config, rank=opt.rank, subprocess_init=subprocess_init)
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/train.py", line 42, in train
    coordinator.train()
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/train_cpu.py", line 701, in train
    epoch_idx, edge_path_idx, edge_chunk_idx, current_index
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/train_cpu.py", line 974, in _maybe_write_checkpoint
    config, self.entity_counts, self.embedding_storage_freelist
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/checkpoint_manager.py", line 407, in write_new_version
    metadata,
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/checkpoint_storage.py", line 322, in save_entity_partition
    with h5py.File(path, "w") as hf:
  File "/root/miniconda3/lib/python3.7/site-packages/h5py/_hl/files.py", line 408, in __init__
    swmr=swmr)
  File "/root/miniconda3/lib/python3.7/site-packages/h5py/_hl/files.py", line 179, in make_fid
    fid = h5f.create(name, h5f.ACC_TRUNC, fapl=fapl, fcpl=fcpl)
  File "h5py/_objects.pyx", line 54, in h5py._objects.with_phil.wrapper
  File "h5py/_objects.pyx", line 55, in h5py._objects.with_phil.wrapper
  File "h5py/h5f.pyx", line 108, in h5py.h5f.create
OSError: Unable to create file (unable to open file: name = '/root/data/release//model/embeddings_schemaorg:MusicRecording_2.v1.h5', errno = 2, error message = 'No such file or directory', flags = 13, o_flags = 242)

it appears that a previous model file is needed.

[lw]

I actually think the issue may be a different one. It seems the trainer is trying to save v1 (i.e., the version produced after the first pass) and thus it didn't fail when it couldn't find v0 (the "prior", if you choose to supply any).

The path that it's trying to write to, however, looks a bit odd: /root/data/release//model/embeddings_schemaorg:MusicRecording_2.v1.h5. There's both the // part (i.e., two separators with no path component between them) and the :. I don't know if either of them could be causing the path to be misinterpreted (if so, my gut feeling is that the former is more likely the culprit), but it would certainly be safer if you could use a "canonical" path with just letters/digits/...

Note that the error says "no such file of directory" during a file creation step, and such an error typically means that one tried to create a file foo/bar but the directory foo/ didn't exist. Could you check if perhaps that is the case too?

[Sherler]

@lw Sorry, I posted a wrong log. The exception during loading stage is the following ones:

Traceback (most recent call last):
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/checkpoint_storage.py", line 341, in load_entity_partition
    with h5py.File(path, "r") as hf:
  File "/root/miniconda3/lib/python3.7/site-packages/h5py/_hl/files.py", line 408, in __init__
    swmr=swmr)
  File "/root/miniconda3/lib/python3.7/site-packages/h5py/_hl/files.py", line 173, in make_fid
    fid = h5f.open(name, flags, fapl=fapl)
  File "h5py/_objects.pyx", line 54, in h5py._objects.with_phil.wrapper
  File "h5py/_objects.pyx", line 55, in h5py._objects.with_phil.wrapper
  File "h5py/h5f.pyx", line 88, in h5py.h5f.open
OSError: Unable to open file (unable to open file: name = '/root/data/release/kgc_full/model/embeddings_schemaorg:BodyOfWater_1.v1.h5', errno = 2, error message = 'No such file or directory', flags = 0, o_flags = 0)

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/root/miniconda3/bin/torchbiggraph_train", line 8, in <module>
    sys.exit(main())
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/train.py", line 71, in main
    train(config, rank=opt.rank, subprocess_init=subprocess_init)
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/train.py", line 42, in train
    coordinator.train()
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/train_cpu.py", line 597, in train
    io_bytes = self._swap_partitioned_embeddings(old_b, cur_b, old_stats)
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/train_cpu.py", line 815, in _swap_partitioned_embeddings
    entity, part, out=embs, strict=self.strict, force_dirty=force_dirty
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/train_cpu.py", line 741, in _load_embeddings
    entity, part, out=out, force_dirty=force_dirty
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/checkpoint_manager.py", line 300, in read
    version, entity, part, out
  File "/root/miniconda3/lib/python3.7/site-packages/torchbiggraph/checkpoint_storage.py", line 350, in load_entity_partition
    raise CouldNotLoadData() from err
torchbiggraph.util.CouldNotLoadData

the directory exists for sure.
the : is ok， as some of the h5 files which contains the : has been created successfully, but some are not (because of the above exception).

accordingly the stdout log of the trainer like this

2020-10-21 09:11:40,760   [Trainer-4] Swapping partitioned embeddings None None
2020-10-21 09:19:29,323   [Trainer-4] Finished epoch 1 / 50, edge path 1 / 1, edge chunk 1 / 1
2020-10-21 09:19:29,324   [Trainer-4] Writing the checkpoint
2020-10-21 09:19:39,077   [Trainer-4] Waiting for other workers to write their parts of the checkpoint
2020-10-21 09:19:49,176   [Trainer-4] All parts of the checkpoint have been written
2020-10-21 09:19:49,176   [Trainer-4] Switching to the new checkpoint version
2020-10-21 09:19:49,177   [Trainer-4] Waiting for other workers to switch to the new checkpoint version
2020-10-21 09:19:49,185   [Trainer-4] All workers have switched to the new checkpoint version
2020-10-21 09:19:49,186   [Trainer-4] Starting epoch 2 / 50, edge path 1 / 1, edge chunk 1 / 1
2020-10-21 09:19:49,186   [Trainer-4] Edge path: /root/data/release/kgc_full/train_partitioned
2020-10-21 09:19:49,203   [Trainer-4] Lock client new epoch...
2020-10-21 09:19:49,257   [Trainer-4] still in queue: 224
2020-10-21 09:19:49,257   [Trainer-4] Swapping partitioned embeddings None ( 1 , 1 )
2020-10-21 09:19:49,257   [Trainer-4] Loading partitioned embeddings from checkpoint

Is there a way to specify save checkpoint only on rank-0 ?

[lw]

What you posted now however is a different error than the previous one: the latter fails when accessing a checkpoint, whereas the former had problems storing a checkpoint. Have you perhaps reused the partial and corrupted checkpoint from the first run when starting the second one? This may explain why it can't find some files that it expects to be there. Could you empty the checkpoint directory and retry?

[Sherler]

This may explain why it can't find some files that it expects to be there. Could you empty the checkpoint directory and retry?

thank you for your reply. Yes, that's a different error. I post the wrong log at first. we can just only focus on the later
I check the PBG source code, After an epoch finished, the checkpoint manager of each rank will just store different part of the model, not all of them.

PyTorch-BigGraph/torchbiggraph/checkpoint_manager.py

Line 389 in 4a58541

for part in range(self.rank, econf.num_partitions, self.num_machines):

As I‘m not using a shared filesystem. Storing partitions partly will cause a file access error, because the trainer needs to load models from the checkpoint_path after Switching to the new checkpoint version, for it cleans the self.dirty set.

PyTorch-BigGraph/torchbiggraph/checkpoint_manager.py

Lines 294 to 299 in 4a58541

	if (entity, part) in self.dirty and self.partition_client is not None:
	embs, serialized_optim_state = self.partition_client.get(entity, part, out)
	else:
	embs, serialized_optim_state = self.storage.load_entity_partition(
	version, entity, part, out
	)

[lw]

I‘m not using a shared filesystem

Oh, well, I hadn't realized this, but this is almost certainly the problem: PBG requires a shared filesystem across trainers when doing distributed training. In practice, it might be possible to get away without it (if you never load a checkpoint and always re-train from scratch), but it's not the way it was intended to be used and thus I don't recommend it and I can't give support for it.