Skip to content

exec 🔗

Exec operations.

ExecMixin 🔗

Mixin providing the Exec operations for a Session.

This mixin is intended to be used in combination with BaseSession and should not be instantiated on its own.

Intended usage

class Session(BaseSession, ExecMixin): ...

exec 🔗

exec(
    library: str | Path,
    symbol: str,
    arg_read: list[Any] | None = None,
    arg_write: list[Any] | None = None,
) -> list[Any] | None

Performs the Exec operation.

Wraps the vaccel_exec() C operation.

Parameters:

Name Type Description Default

library 🔗

 str | Path

The path to the shared object containing the function that will be called.

 required

symbol 🔗

 str

The name of the function contained in the above shared object.

 required

arg_read 🔗

 list[Any] | None

The input arguments that will be passed to the called function.

 None

arg_write 🔗

 list[Any] | None

The output arguments that will be passed to the called function.

 None

Returns:

Type Description
list[Any] | None

The resulting outputs.

Raises:

Type Description
FFIError

If the C operation fails.
Source code in vaccel/ops/exec.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def exec(
    self,
    library: str | Path,
    symbol: str,
    arg_read: list[Any] | None = None,
    arg_write: list[Any] | None = None,
) -> list[Any] | None:
    """Performs the Exec operation.

    Wraps the `vaccel_exec()` C operation.

    Args:
        library: The path to the shared object containing the function that
            will be called.
        symbol: The name of the function contained in the above shared
            object.
        arg_read: The input arguments that will be passed to the
            called function.
        arg_write: The output arguments that will be passed to the
            called function.

    Returns:
        The resulting outputs.

    Raises:
        FFIError: If the C operation fails.
    """
    if arg_read is not None:
        c_arg_read = CList([Arg(arg) for arg in arg_read])
        c_arg_read_ptr = c_arg_read._c_ptr
        c_arg_read_len = len(c_arg_read)
    else:
        c_arg_read = None
        c_arg_read_ptr = ffi.NULL
        c_arg_read_len = 0

    if arg_write is not None:
        c_arg_write = CList([Arg(arg) for arg in arg_write])
        c_arg_write_ptr = c_arg_write._c_ptr
        c_arg_write_len = len(c_arg_write)
    else:
        c_arg_write = None
        c_arg_write_ptr = ffi.NULL
        c_arg_write_len = 0

    ret = lib.vaccel_exec(
        self._c_ptr_or_raise,
        str(library).encode(),
        symbol.encode(),
        c_arg_read_ptr,
        c_arg_read_len,
        c_arg_write_ptr,
        c_arg_write_len,
    )
    if ret != 0:
        raise FFIError(ret, "Exec operation failed")

    if c_arg_write is not None:
        return [c_arg_write[i].buf for i in range(len(c_arg_write))]
    return None

exec_with_resource 🔗

exec_with_resource(
    resource: Resource,
    symbol: str,
    arg_read: list[Any] | None = None,
    arg_write: list[Any] | None = None,
) -> list[Any] | None

Performs the Exec with resource operation.

Wraps the vaccel_exec_with_resource() C operation.

Parameters:

Name Type Description Default

resource 🔗

 Resource

The resource of the shared object containing the function that will be called.

 required

symbol 🔗

 str

The name of the function contained in the above shared object.

 required

arg_read 🔗

 list[Any] | None

The input arguments that will be passed to the called function.

 None

arg_write 🔗

 list[Any] | None

The output arguments that will be passed to the called function.

 None

Returns:

Type Description
list[Any] | None

The resulting outputs.

Raises:

Type Description
FFIError

If the C operation fails.
Source code in vaccel/ops/exec.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def exec_with_resource(
    self,
    resource: Resource,
    symbol: str,
    arg_read: list[Any] | None = None,
    arg_write: list[Any] | None = None,
) -> list[Any] | None:
    """Performs the Exec with resource operation.

    Wraps the `vaccel_exec_with_resource()` C operation.

    Args:
        resource: The resource of the shared object containing the function
            that will be called.
        symbol: The name of the function contained in the above shared
            object.
        arg_read: The input arguments that will be passed to the
            called function.
        arg_write: The output arguments that will be passed to the
            called function.

    Returns:
        The resulting outputs.

    Raises:
        FFIError: If the C operation fails.
    """
    if arg_read is not None:
        c_arg_read = CList([Arg(arg) for arg in arg_read])
        c_arg_read_ptr = c_arg_read._c_ptr
        c_arg_read_len = len(c_arg_read)
    else:
        c_arg_read = None
        c_arg_read_ptr = ffi.NULL
        c_arg_read_len = 0

    if arg_write is not None:
        c_arg_write = CList([Arg(arg) for arg in arg_write])
        c_arg_write_ptr = c_arg_write._c_ptr
        c_arg_write_len = len(c_arg_write)
    else:
        c_arg_write = None
        c_arg_write_ptr = ffi.NULL
        c_arg_write_len = 0

    ret = lib.vaccel_exec_with_resource(
        self._c_ptr_or_raise,
        resource._c_ptr,
        symbol.encode(),
        c_arg_read_ptr,
        c_arg_read_len,
        c_arg_write_ptr,
        c_arg_write_len,
    )
    if ret != 0:
        raise FFIError(ret, "Exec with resource operation failed")

    if c_arg_write is not None:
        return [c_arg_write[i].buf for i in range(len(c_arg_write))]
    return None