Protocol Implementation

We've seen in the earlier section how a PulseAudio application, using the PA client libraries, locate the PA daemon.

We now discuss the connection protocol details after initiating the connection with the daemon. What does the protocol look like?

On connection

Upon succeeding in finding an appropriate socket for communication with the daemon, the PA client libraries code registers the callback on_connection using PA IO event loop mechanisms. It exists at src/pulse/context.c:

static void on_connection(pa_socket_client *client, pa_iochannel*io, void *userdata) {
    pa_context *c = userdata;
    pa_assert(c->state == PA_CONTEXT_CONNECTING);
    ...
    setup_context(c, io);
}

Now on the same file, setup_context() does:

static void setup_context(pa_context *c, pa_iochannel *io) {
    uint8_t cookie[PA_NATIVE_COOKIE_LENGTH];
    pa_tagstruct *t;
    uint32_t tag;

    c->pstream = pa_pstream_new(c->mainloop, io, c->mempool);

    pa_pstream_set_die_callback(c->pstream, pstream_die_callback, c);
    pa_pstream_set_receive_packet_callback(c->pstream, pstream_packet_callback, c);
    pa_pstream_set_receive_memblock_callback(c->pstream, pstream_memblock_callback, c);

    pa_assert(!c->pdispatch);
    c->pdispatch = pa_pdispatch_new(c->mainloop, c->use_rtclock, command_table, PA_COMMAND_MAX);

    if (pa_client_conf_load_cookie(c->conf, cookie, sizeof(cookie)) < 0)
        pa_log_info("No cookie loaded. Attempting to connect without.");

    t = pa_tagstruct_command(c, PA_COMMAND_AUTH, &tag);

    c->do_shm =
        pa_mempool_is_shared(c->mempool) &&
        c->is_local;

    pa_log_debug("SHM possible: %s", pa_yes_no(c->do_shm));

    /* Starting with protocol version 13 we use the MSB of the version
     * tag for informing the other side if we could do SHM or not */
    pa_tagstruct_putu32(t, PA_PROTOCOL_VERSION | (c->do_shm ? 0x80000000U : 0));
    pa_tagstruct_put_arbitrary(t, cookie, sizeof(cookie));

    pa_pstream_send_tagstruct(c->pstream, t);

    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, c, NULL);

    pa_context_set_state(c, PA_CONTEXT_AUTHORIZING);

    pa_context_unref(c);
}

What we see above is the PulseAudio library code registering multiple callbacks.

Responding to packets: pstream_packet_callback()

static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, const pa_cmsg_ancil_data *ancil_data, void *userdata) {
    pa_context *c = userdata;

    if (pa_pdispatch_run(c->pdispatch, packet, ancil_data, c) < 0)
        pa_context_fail(c, PA_ERR_PROTOCOL);
}

This basically runs PA commands from the commands vtable. To show all the commands received from both server and client sides, define DEBUG_OPCODES on top of src/pulsecore/pdispatch.c

Responding to memblock requests: pstream_memblock_callback

static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, int64_t offset, pa_seek_mode_t seek, const pa_memchunk *chunk, void *userdata) {
    pa_context *c = userdata;
    pa_stream *s;

    if ((s = pa_hashmap_get(c->record_streams, PA_UINT32_TO_PTR(channel)))) {
        ...
    }

    pa_context_unref(c);
}

This is mostly for recording applications.

Sample case of commands sent and received

Define DEBUG_OPCODES on top of src/pulsecore/pdispatch.c, then do the following:

$ LD_LIBRARY_PATH=$PULSE_SRC_DIR/build/src/.libs/ PULSE_LOG=99 PULSE_LOG_COLORS= $PULSE_SRC_DIR/build/src/pactl info

Parsing configuration file '/home/darwish/.pulse/client.conf'
Using shared memory pool with 1024 slots of size 64.0 KiB each, total size is 64.0 MiB, maximum usable slot size is 65472
** Adding to server list: /home/darwish/playground/pulseaudio/build/var/run/pulse/native
** Adding to server list: /run/user/1000/pulse/native
Trying to connect to /run/user/1000/pulse/native...
SHM possible: yes
[0x15f0cb0] Received opcode <REPLY>
Protocol version: remote 30, local 30
Negotiated SHM: yes
[0x15f0cb0] Received opcode <ENABLE_SRBCHANNEL>
[0x15f0cb0] Received opcode <REPLY>
[0x15f0cb0] _Sending_ opcode <GET_SERVER_INFO>
[0x15f0cb0] Received opcode <REPLY>
... Then server information follows ...

And while inspecting the daemon log in the same moment, we have:

I: [lt-pulseaudio] client.c: Created 7 "Native client (UNIX socket client)"
E: [lt-pulseaudio] pdispatch.c: [0x2441c20] Received opcode <AUTH>
D: [lt-pulseaudio] protocol-native.c: Protocol version: remote 30, local 30
I: [lt-pulseaudio] protocol-native.c: Got credentials: uid=1000 gid=1000 success=1
D: [lt-pulseaudio] protocol-native.c: SHM possible: yes
D: [lt-pulseaudio] protocol-native.c: Negotiated SHM: yes
D: [lt-pulseaudio] srbchannel.c: SHM block is 65472 bytes, ringbuffer capacity is 2 * 32712 bytes
D: [lt-pulseaudio] protocol-native.c: Enabling srbchannel...
E: [lt-pulseaudio] pdispatch.c: [0x2424a00] Received opcode <GET_CLIENT_INFO>
E: [lt-pulseaudio] pdispatch.c: [0x2441c20] Received opcode <SET_CLIENT_NAME>
D: [lt-pulseaudio] module-augment-properties.c: Looking for .desktop file for lt-pactl
E: [lt-pulseaudio] pdispatch.c: [0x2441c20] Received opcode <ENABLE_SRBCHANNEL>
D: [lt-pulseaudio] protocol-native.c: Client enabled srbchannel.
E: [lt-pulseaudio] pdispatch.c: [0x2441c20] Received opcode <GET_SERVER_INFO>
E: [lt-pulseaudio] pdispatch.c: [0x2424a00] Received opcode <GET_CLIENT_INFO>
I: [lt-pulseaudio] client.c: Freed 7 "lt-pactl"
I: [lt-pulseaudio] protocol-native.c: Connection died.

Let's inspect the commands pactl sends to the PA daemon: -

• GET_SERVER_INFO is pretty explanatory. What's pretty intruguing is the ENABLE_SRBCHANNEL command.

• ENABLE_SRCHANNEL stands for Shared-memory Ringbuffer Channel. It's an update to the PulseAudio protocol to use less system calls in the communication from the client to the daemon and back. This is especially necessary for low-latency loads like Skype or VOIP. Check our PA 6.0 Protocol section for further details.

• REPLY: From inspecting the daemon and client libraries log, this is only sent from the daemon to the client. But what does such command imply? It's a simple ACK mechanism to reply with when we receive a command that does not expect a reply value.

Further notes on the REPLY command

For further evience, we can check src/pulsecore/pstream-util.c:

void pa_pstream_send_simple_ack(pa_pstream *p, uint32_t tag) {
    pa_tagstruct *t;

    pa_assert_se(t = pa_tagstruct_new());
    pa_tagstruct_putu32(t, PA_COMMAND_REPLY);
    pa_tagstruct_putu32(t, tag);
    pa_pstream_send_tagstruct(p, t);
}

And as can be seen, the simple ACK mechanism is to send the very simple REPLY command.

Also, we can see a very high amount of usage of pa_pstream_send_simple_ack in the implementation of different PulseAudio commands under src/pulsecore/protocol-native.c. The commands that sends a REPLY are usually commands which:

• have nothing to return, thus they return REPLY as a success mark
• or commands which when succeeds, have nothing to return (thus returning REPLY). But on failure, they return an error.

static const pa_pdispatch_cb_t command_table[PA_COMMAND_MAX] = {
    [PA_COMMAND_ERROR] = NULL,
    [PA_COMMAND_TIMEOUT] = NULL,
    [PA_COMMAND_REPLY] = NULL,
    ...
    [PA_COMMAND_SET_SOURCE_OUTPUT_VOLUME] = command_set_volume,
    [PA_COMMAND_SET_SINK_MUTE] = command_set_mute,
    [PA_COMMAND_SUSPEND_SINK] = command_suspend,
    ...
    [PA_COMMAND_SET_PLAYBACK_STREAM_NAME] = command_set_stream_name,
    [PA_COMMAND_KILL_CLIENT] = command_kill,
    [PA_COMMAND_KILL_SINK_INPUT] = command_kill,

};

And we can see that the implementation of all of the above commands does not return a value, and thus sends a simple REPLY command at the end.

static void command_kill(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
    pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
    uint32_t idx;
    if (pa_tagstruct_getu32(t, &idx) < 0 ||
        !pa_tagstruct_eof(t)) {
        protocol_error(c);
        return;
    }

    if (command == PA_COMMAND_KILL_CLIENT) {
        ...
    } else {
        ...
    }

    pa_pstream_send_simple_ack(c->pstream, tag);
}

static void command_suspend(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
    pa_native_connection *c = PA_NATIVE_CONNECTION(userdata);
    uint32_t idx = PA_INVALID_INDEX;
    const char *name = NULL;
    bool b;

    if (pa_tagstruct_getu32(t, &idx) < 0 ||
        pa_tagstruct_gets(t, &name) < 0 ||
        pa_tagstruct_get_boolean(t, &b) < 0 ||
        !pa_tagstruct_eof(t)) {
        protocol_error(c);
        return;
    }

    if (command == PA_COMMAND_SUSPEND_SINK) {
        ...
    } else {
        pa_assert(command == PA_COMMAND_SUSPEND_SOURCE);
        ...
    } else {
        ...
    }

    pa_pstream_send_simple_ack(c->pstream, tag);
}