| 395 | raise Exception(f"Failed to write to infinistore, ret = {ret}") |
| 396 | |
| 397 | async def rdma_write_cache_async( |
| 398 | self, blocks: List[Tuple[str, int]], block_size: int, ptr: int |
| 399 | ): |
| 400 | if not self.rdma_connected: |
| 401 | raise Exception("this function is only valid for connected rdma") |
| 402 | |
| 403 | await self.semaphore.acquire() |
| 404 | loop = asyncio.get_running_loop() |
| 405 | future = loop.create_future() |
| 406 | |
| 407 | keys, offsets = zip(*blocks) |
| 408 | |
| 409 | def _callback(code): |
| 410 | if code != 200: |
| 411 | loop.call_soon_threadsafe( |
| 412 | future.set_exception, |
| 413 | Exception(f"Failed to write to infinistore, ret = {code}"), |
| 414 | ) |
| 415 | else: |
| 416 | loop.call_soon_threadsafe(future.set_result, code) |
| 417 | self.semaphore.release() |
| 418 | |
| 419 | ret = self.conn.w_rdma_async( |
| 420 | keys, |
| 421 | offsets, |
| 422 | block_size, |
| 423 | ptr, |
| 424 | _callback, |
| 425 | ) |
| 426 | if ret < 0: |
| 427 | raise Exception(f"Failed to write to infinistore, ret = {ret}") |
| 428 | return await future |
| 429 | |
| 430 | async def rdma_read_cache_async( |
| 431 | self, blocks: List[Tuple[str, int]], block_size: int, ptr: int |