(self, server_state)
| 565 | ) = state |
| 566 | |
| 567 | def process_request(self, server_state): |
| 568 | kv_store = server_state.kv_store |
| 569 | assert kv_store.is_backup_server() |
| 570 | if self.name not in kv_store.data_store: |
| 571 | shared_data = empty_shared_mem( |
| 572 | self.name + "-kvdata-", False, self.shape, self.dtype |
| 573 | ) |
| 574 | dlpack = shared_data.to_dlpack() |
| 575 | kv_store.data_store[self.name] = F.zerocopy_from_dlpack(dlpack) |
| 576 | kv_store.part_policy[self.name] = kv_store.find_policy( |
| 577 | self.policy_str |
| 578 | ) |
| 579 | kv_store.pull_handlers[self.name] = self.pull_handler |
| 580 | kv_store.push_handlers[self.name] = self.push_handler |
| 581 | else: |
| 582 | assert tuple(F.shape(kv_store.data_store[self.name])) == tuple( |
| 583 | self.shape |
| 584 | ) |
| 585 | assert ( |
| 586 | F.reverse_data_type_dict[ |
| 587 | F.dtype(kv_store.data_store[self.name]) |
| 588 | ] |
| 589 | == self.dtype |
| 590 | ) |
| 591 | assert kv_store.part_policy[self.name].policy_str == self.policy_str |
| 592 | assert kv_store.pull_handlers[self.name] == self.pull_handler |
| 593 | assert kv_store.push_handlers[self.name] == self.push_handler |
| 594 | res = SendMetaToBackupResponse(SEND_META_TO_BACKUP_MSG) |
| 595 | return res |
| 596 | |
| 597 | |
| 598 | DELETE_DATA = 901240 |
nothing calls this directly
no test coverage detected