GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2025-12-12 09:10:30 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Issue pikvm/pikvm#1235: Fixed gadgets on UDC re-bind

Browse Source 
After unbind and bind, the gadgets stop working,
unless we recreate their links in the profile.
Some kind of kernel bug.
This commit is contained in:
Maxim Devaev 2024-02-05 16:04:28 +02:00
parent d3f2b57fdc
commit af9c2f1f59
2 changed files with 56 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
kvmd/apps/otgconf/__init__.py
View File

@ -57,31 +57,51 @@ class _GadgetControl:
try:
yield
finally:
self.__recreate_profile()
time.sleep(self.__init_delay)
with open(udc_path, "w") as file:
file.write(udc)
def __recreate_profile(self) -> None:
# XXX: See pikvm/pikvm#1235
# After unbind and bind, the gadgets stop working,
# unless we recreate their links in the profile.
# Some kind of kernel bug.
for func in os.listdir(self.__get_fdest_path()):
path = self.__get_fdest_path(func)
if os.path.islink(path):
try:
os.unlink(path)
os.symlink(self.__get_fsrc_path(func), path)
except (FileNotFoundError, FileExistsError):
pass
def __read_metas(self) -> Generator[dict, None, None]:
for meta_name in sorted(os.listdir(self.__meta_path)):
with open(os.path.join(self.__meta_path, meta_name)) as file:
yield json.loads(file.read())
def __get_fsrc_path(self, func: str) -> str:
return usb.get_gadget_path(self.__gadget, usb.G_FUNCTIONS, func)
def __get_fdest_path(self, func: (str | None)=None) -> str:
if func is None:
return usb.get_gadget_path(self.__gadget, usb.G_PROFILE)
return usb.get_gadget_path(self.__gadget, usb.G_PROFILE, func)
def enable_functions(self, funcs: list[str]) -> None:
with self.__udc_stopped():
for func in funcs:
os.symlink(
usb.get_gadget_path(self.__gadget, usb.G_FUNCTIONS, func),
usb.get_gadget_path(self.__gadget, usb.G_PROFILE, func),
)
os.symlink(self.__get_fsrc_path(func), self.__get_fdest_path(func))
def disable_functions(self, funcs: list[str]) -> None:
with self.__udc_stopped():
for func in funcs:
os.unlink(usb.get_gadget_path(self.__gadget, usb.G_PROFILE, func))
os.unlink(self.__get_fdest_path(func))
def list_functions(self) -> None:
for meta in self.__read_metas():
enabled = os.path.exists(usb.get_gadget_path(self.__gadget, usb.G_PROFILE, meta["func"]))
enabled = os.path.exists(self.__get_fdest_path(meta["func"]))
print(f"{'+' if enabled else '-'} {meta['func']} # {meta['name']}")
def make_gpio_config(self) -> None:

36
kvmd/plugins/ugpio/otgconf.py
View File

@ -105,29 +105,53 @@ class Plugin(BaseUserGpioDriver):
async def read(self, pin: str) -> bool:
if pin == "udc":
return self.__is_udc_enabled()
return os.path.exists(os.path.join(self.__profile_path, pin))
return os.path.exists(self.__get_fdest_path(pin))
async def write(self, pin: str, state: bool) -> None:
async with self.__lock:
if self.read(pin) == state:
return
if pin == "udc":
if state:
self.__recreate_profile()
self.__set_udc_enabled(state)
else:
if self.__is_udc_enabled():
self.__set_udc_enabled(False)
try:
if state:
os.symlink(
os.path.join(self.__functions_path, pin),
os.path.join(self.__profile_path, pin),
)
os.symlink(self.__get_fsrc_path(pin), self.__get_fdest_path(pin))
else:
os.unlink(os.path.join(self.__profile_path, pin))
os.unlink(self.__get_fdest_path(pin))
except (FileNotFoundError, FileExistsError):
pass
finally:
self.__recreate_profile()
try:
await asyncio.sleep(self.__init_delay)
finally:
self.__set_udc_enabled(True)
def __recreate_profile(self) -> None:
# XXX: See pikvm/pikvm#1235
# After unbind and bind, the gadgets stop working,
# unless we recreate their links in the profile.
# Some kind of kernel bug.
for func in os.listdir(self.__profile_path):
path = self.__get_fdest_path(func)
if os.path.islink(path):
try:
os.unlink(path)
os.symlink(self.__get_fsrc_path(func), path)
except (FileNotFoundError, FileNotFoundError):
pass
def __get_fsrc_path(self, func: str) -> str:
return os.path.join(self.__functions_path, func)
def __get_fdest_path(self, func: str) -> str:
return os.path.join(self.__profile_path, func)
def __set_udc_enabled(self, enabled: bool) -> None:
with open(self.__udc_path, "w") as file:
file.write(self.__udc if enabled else "\n")