matthew/sonic-buildimage
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
master
sonic-buildimage/device/celestica/x86_64-cel_seastone-r0/sonic_platform/component.py
Ikki Zhu 9fcbd5ed1d
fix possible cpld race access issue (#15371) 
Why I did it
fix possible cpld race read issue between watchdog and reboot cause
process

How I did it
Use fcntl.flock to limit parallel access to cpld sys file

How to verify it
It can be simulated and verified with following python script

``` python3
import fcntl
import signal
import threading

exit_flag = False

def get_cpld_reg_value(getreg_path, register):
    file = open(getreg_path, 'w+')
    # Acquire an exclusive lock on the file
    fcntl.flock(file, fcntl.LOCK_EX)

    try:
        file.write(register + '\n')
        file.flush()

        # Seek to the beginning of the file
        file.seek(0)

        # Read the content of the file
        result = file.readline().strip()
    finally:
        # Release the lock and close the file
        fcntl.flock(file, fcntl.LOCK_UN)
        file.close()

    return result

def cpld_read(thread_num, cpld_reg, expect_val):
    while not exit_flag:
        val
= get_cpld_reg_value("/sys/devices/platform/dx010_cpld/getreg",
cpld_reg)
        #print(f"Thread {thread_num}: get cpld reg {cpld_reg}, value
{val}")
        if val != expect_val:
            print(f"Thread {thread_num}: get cpld reg {cpld_reg}, value
{val}, expect_val {expect_val}")

def signal_handler(sig, frame):
    global exit_flag
    print("Ctrl+C detected. Quitting...")
    exit_flag = True

if __name__ == '__main__':
    # Register the signal handler for Ctrl+C
    signal.signal(signal.SIGINT, signal_handler)

    t1 = threading.Thread(target=cpld_read, args=(1, '0x103', '0x11',))
    t2 = threading.Thread(target=cpld_read, args=(2, '0x141', '0x00',))
    t1.start()
    t2.start()
    t1.join()
    t2.join()
```
2023-06-07 11:29:18 -07:00

217 lines
7.2 KiB
Python
Raw Permalink Blame History

#############################################################################
# Celestica
#
# Component contains an implementation of SONiC Platform Base API and
# provides the components firmware management function
#
#############################################################################
import os.path
import shutil
try:
from sonic_platform_base.component_base import ComponentBase
from .helper import APIHelper
except ImportError as e:
raise ImportError(str(e) + "- required module not found")
CPLD_ADDR_MAPPING = {
"CPLD1": "0x100",
"CPLD2": "0x200",
"CPLD3": "0x280",
"CPLD4": "0x300",
"CPLD5": "0x380"
}
GETREG_PATH = "/sys/devices/platform/dx010_cpld/getreg"
BIOS_VERSION_PATH = "/sys/class/dmi/id/bios_version"
COMPONENT_NAME_LIST = ["CPLD1", "CPLD2", "CPLD3", "CPLD4", "CPLD5", "BIOS"]
COMPONENT_DES_LIST = ["Used for managing the CPU",
"Used for managing QSFP+ ports (1-10)", "Used for managing QSFP+ ports (11-21)",
"Used for misc status and control", "Used for managing QSFP+ ports (22-32)",
"Basic Input/Output System"]
class Component(ComponentBase):
"""Platform-specific Component class"""
DEVICE_TYPE = "component"
def __init__(self, component_index):
ComponentBase.__init__(self)
self.index = component_index
self._api_helper = APIHelper()
self.name = self.get_name()
def __get_bios_version(self):
# Retrieves the BIOS firmware version
try:
with open(BIOS_VERSION_PATH, 'r') as fd:
bios_version = fd.read()
return bios_version.strip()
except Exception as e:
return None
def __get_cpld_version(self):
# Retrieves the CPLD firmware version
cpld_version = dict()
for cpld_name in CPLD_ADDR_MAPPING:
try:
cpld_addr = CPLD_ADDR_MAPPING[cpld_name]
cpld_version_raw = self._api_helper.get_cpld_reg_value(GETREG_PATH, cpld_addr)
cpld_version_str = "{}.{}".format(int(cpld_version_raw[2], 16), int(
cpld_version_raw[3], 16)) if cpld_version_raw is not None else 'None'
cpld_version[cpld_name] = cpld_version_str
except Exception as e:
cpld_version[cpld_name] = 'None'
return cpld_version
def get_name(self):
"""
Retrieves the name of the component
Returns:
A string containing the name of the component
"""
return COMPONENT_NAME_LIST[self.index]
def get_description(self):
"""
Retrieves the description of the component
Returns:
A string containing the description of the component
"""
return COMPONENT_DES_LIST[self.index]
def get_firmware_version(self):
"""
Retrieves the firmware version of module
Returns:
string: The firmware versions of the module
"""
fw_version = None
if self.name == "BIOS":
fw_version = self.__get_bios_version()
elif "CPLD" in self.name:
cpld_version = self.__get_cpld_version()
fw_version = cpld_version.get(self.name)
return fw_version
def get_available_firmware_version(self, image_path):
"""
Retrieves the available firmware version of the component
Note: the firmware version will be read from image
Args:
image_path: A string, path to firmware image
Returns:
A string containing the available firmware version of the component
"""
return "N/A"
def get_firmware_update_notification(self, image_path):
"""
Retrieves a notification on what should be done in order to complete
the component firmware update
Args:
image_path: A string, path to firmware image
Returns:
A string containing the component firmware update notification if required.
By default 'None' value will be used, which indicates that no actions are required
"""
return "None"
def install_firmware(self, image_path):
"""
Install firmware to module
Args:
image_path: A string, path to firmware image
Returns:
A boolean, True if install successfully, False if not
"""
if not os.path.isfile(image_path):
return False
if "CPLD" in self.name:
img_name = os.path.basename(image_path)
root, ext = os.path.splitext(img_name)
ext = ".vme" if ext == "" else ext
new_image_path = os.path.join("/tmp", (root.lower() + ext))
shutil.copy(image_path, new_image_path)
install_command = ["ispvm", str(new_image_path)]
# elif self.name == "BIOS":
# install_command = "afulnx_64 %s /p /b /n /x /r" % image_path
return self._api_helper.run_command(install_command)
def update_firmware(self, image_path):
"""
Updates firmware of the component
This API performs firmware update: it assumes firmware installation and loading in a single call.
In case platform component requires some extra steps (apart from calling Low Level Utility)
to load the installed firmware (e.g, reboot, power cycle, etc.) - this will be done automatically by API
Args:
image_path: A string, path to firmware image
Raises:
RuntimeError: update failed
"""
return False
##############################################################
###################### Device methods ########################
##############################################################
def get_presence(self):
"""
Retrieves the presence of the FAN
Returns:
bool: True if FAN is present, False if not
"""
return True
def get_model(self):
"""
Retrieves the model number (or part number) of the device
Returns:
string: Model/part number of device
"""
return 'N/A'
def get_serial(self):
"""
Retrieves the serial number of the device
Returns:
string: Serial number of device
"""
return 'N/A'
def get_status(self):
"""
Retrieves the operational status of the device
Returns:
A boolean value, True if device is operating properly, False if not
"""
return True
def get_position_in_parent(self):
"""
Retrieves 1-based relative physical position in parent device.
If the agent cannot determine the parent-relative position
for some reason, or if the associated value of
entPhysicalContainedIn is'0', then the value '-1' is returned
Returns:
integer: The 1-based relative physical position in parent device
or -1 if cannot determine the position
"""
return -1
def is_replaceable(self):
"""
Indicate whether this device is replaceable.
Returns:
bool: True if it is replaceable.
"""
return False
Reference in New Issue View Git Blame Copy Permalink