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], arg_write: list[Any]
) -> list[Any]

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]

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

 required

arg_write 🔗

 list[Any]

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

 required

Returns:

Type Description
list[Any]

The resulting outputs.

Raises:

Type Description
RuntimeError

If the Session is uninitialized.
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
def exec(
    self,
    library: str | Path,
    symbol: str,
    arg_read: list[Any],
    arg_write: list[Any],
) -> list[Any]:
    """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:
        RuntimeError: If the `Session` is uninitialized.
        FFIError: If the C operation fails.
    """
    if not self._c_ptr:
        msg = "Uninitialized session"
        raise RuntimeError(msg)

    c_arg_read = CList([Arg(arg) for arg in arg_read])
    c_arg_write = CList([Arg(arg) for arg in arg_write])

    ret = lib.vaccel_exec(
        self._c_ptr,
        str(library).encode(),
        symbol.encode(),
        c_arg_read._c_ptr,
        len(c_arg_read),
        c_arg_write._c_ptr,
        len(c_arg_write),
    )
    if ret != 0:
        raise FFIError(ret, "Exec operation failed")

    return [c_arg_write[i].buf for i in range(len(c_arg_write))]

exec_with_resource 🔗

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

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]

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

 required

arg_write 🔗

 list[Any]

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

 required

Returns:

Type Description
list[Any]

The resulting outputs.

Raises:

Type Description
RuntimeError

If the Session is uninitialized.
FFIError

If the C operation fails.
Source code in vaccel/ops/exec.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 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
def exec_with_resource(
    self,
    resource: Resource,
    symbol: str,
    arg_read: list[Any],
    arg_write: list[Any],
) -> list[Any]:
    """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:
        RuntimeError: If the `Session` is uninitialized.
        FFIError: If the C operation fails.
    """
    if not self._c_ptr:
        msg = "Uninitialized session"
        raise RuntimeError(msg)

    c_arg_read = CList([Arg(arg) for arg in arg_read])
    c_arg_write = CList([Arg(arg) for arg in arg_write])

    ret = lib.vaccel_exec_with_resource(
        self._c_ptr,
        resource._c_ptr,
        symbol.encode(),
        c_arg_read._c_ptr,
        len(c_arg_read),
        c_arg_write._c_ptr,
        len(c_arg_write),
    )
    if ret != 0:
        raise FFIError(ret, "Exec with resource operation failed")

    return [c_arg_write[i].buf for i in range(len(c_arg_write))]