Truncate file through EOS using XRootD File interface

src/fsspec_xrootd/xrootd.py

@@ -1,4 +1,4 @@
 from __future__ import annotations
 
 import asyncio
 import io
@@ -15,18 +15,19 @@
 from fsspec.asyn import AsyncFileSystem, _run_coros_in_chunks, sync, sync_wrapper
 from fsspec.exceptions import FSTimeoutError
 from fsspec.spec import AbstractBufferedFile
 from XRootD import client
+from XRootD.client import File
 from XRootD.client.flags import (
     DirListFlags,
     MkDirFlags,
     OpenFlags,
     QueryCode,
     StatInfoFlags,
 )
 from XRootD.client.responses import HostList, XRootDStatus
 
 
 class ErrorCodes(IntEnum):
     INVALID_PATH = 400
 
 
@@ -51,13 +52,13 @@
         asyncio.get_running_loop().create_future()
     )
 
     def callback(status: XRootDStatus, content: T, servers: HostList) -> None:
         if future.cancelled():
             return
         loop = future.get_loop()
         try:
             loop.call_soon_threadsafe(future.set_result, (status, content))
         except Exception as exc:
             loop.call_soon_threadsafe(future.set_exception, exc)
 
     async def wrapped(*args: Any, **kwargs: Any) -> tuple[XRootDStatus, T]:
@@ -142,7 +143,7 @@
     handle: client.File
 
 
 class ReadonlyFileHandleCache:
     def __init__(self, loop: Any, max_items: int | None, ttl: int):
         self.loop = loop
         self._max_items = max_items
@@ -170,7 +171,7 @@
                 item.handle.close(callback=lambda *args: None)
         cache.clear()
 
     def close_all(self) -> None:
         self._close_all(self.loop, self._cache)
 
     async def _close(self, url: str, timeout: int) -> None:
@@ -183,7 +184,7 @@
     close = sync_wrapper(_close)
 
     async def _start_pruner(self) -> None:
         self._prune_task = asyncio.create_task(self._pruner())
 
     async def _pruner(self) -> None:
         while True:
@@ -220,7 +221,7 @@
         return handle
 
 
 class XRootDFileSystem(AsyncFileSystem):  # type: ignore[misc]
     protocol = "root"
     root_marker = "/"
     default_timeout = 60
@@ -372,21 +373,36 @@
             raise OSError(f"File not removed properly: {status.message}")
 
     async def _touch(self, path: str, truncate: bool = False, **kwargs: Any) -> None:
-        if truncate or not await self._exists(path):
-            status, _ = await _async_wrap(self._myclient.truncate)(
-                path, size=0, timeout=self.timeout
-            )
+        # necessary to write in EOS, see https://github.com/xrootd/xrootd/issues/2304
+        if "eos" in self.hostid:
+            eos_full_path = f"{self.protocol}://{self.hostid}/{path}"
+            f = File()
+            status, _ = f.open(eos_full_path, OpenFlags.NEW, timeout=self.timeout)
             if not status.ok:
-                raise OSError(f"File not touched properly: {status.message}")
+                raise OSError(f"Impossible to create file in EOS: {status.message}")
+            if truncate or not await self._exists(path):
+                status, _ = await _async_wrap(f.truncate)(size=0, timeout=self.timeout)
+            else:
+                len = await self._info(path)
+                status, _ = await _async_wrap(f.truncate)(
+                    size=len.get("size"),
+                    timeout=self.timeout,
+                )
+            f.close()
         else:
-            len = await self._info(path)
-            status, _ = await _async_wrap(self._myclient.truncate)(
-                path,
-                size=len.get("size"),
-                timeout=self.timeout,
-            )
-            if not status.ok:
-                raise OSError(f"File not touched properly: {status.message}")
+            if truncate or not await self._exists(path):
+                status, _ = await _async_wrap(self._myclient.truncate)(
+                    path, size=0, timeout=self.timeout
+                )
+            else:
+                len = await self._info(path)
+                status, _ = await _async_wrap(self._myclient.truncate)(
+                    path,
+                    size=len.get("size"),
+                    timeout=self.timeout,
+                )
+        if not status.ok:
+            raise OSError(f"File not touched properly: {status.message}")
 
     touch = sync_wrapper(_touch)
 
@@ -834,7 +850,7 @@
 
         self.kwargs = kwargs
 
-        if mode not in {"ab", "rb", "wb"}:
+        if mode not in {"ab", "rb", "wb", "a+b", "r+b", "w+b"}:
             raise NotImplementedError("File mode not supported")
         if mode == "rb":
             if size is not None: