diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..6b8710a --- /dev/null +++ b/.dockerignore @@ -0,0 +1 @@ +.git diff --git a/.github/workflows/qa.yml b/.github/workflows/qa.yml new file mode 100644 index 0000000..61a8e7e --- /dev/null +++ b/.github/workflows/qa.yml @@ -0,0 +1,25 @@ +name: qa + +on: + push: + branches: [questnet] + pull_request: + workflow_dispatch: + +# a new push to the same branch or PR cancels the runs still in progress +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + qa: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + # the two configurations of `just test-all`: scalar vector math, and SSE2 (runners are linux/amd64) + target: [test, test-sse2] + steps: + - uses: actions/checkout@v4 + - uses: extractions/setup-just@v2 + - run: just ${{ matrix.target }} diff --git a/.gitignore b/.gitignore index 259148f..2f83e50 100644 --- a/.gitignore +++ b/.gitignore @@ -30,3 +30,8 @@ *.exe *.out *.app + +# Local IDE support (just ide-setup) +.deps/ +.clangd +.vscode/c_cpp_properties.json diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..ddcb509 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,5 @@ +# Notes for agents + +- Periodically check the upstream repositories listed in [issues/upstream-watch.md](issues/upstream-watch.md) for changes that affect `mod_audio_fork`, and update that file when you do. If the user asks about upstream changes, start there. +- Open review findings and what was already fixed: [issues/review-findings.md](issues/review-findings.md). +- Build and test: `just test` (runs in Docker). diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..f50b085 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,56 @@ +# Compile-only check of mod_audio_fork against FreeSWITCH headers and libwebsockets, +# plus standalone unit tests. Nothing is linked against libfreeswitch. +# docker build . +# docker build --platform linux/amd64 --build-arg SIMD=sse2 . +FROM debian:trixie-slim + +ARG FS_VERSION=v1.11.3 +ARG LWS_VERSION=v5.0.0 +ARG SIMD=none + +RUN apt-get update && apt-get install -y --no-install-recommends \ + build-essential cmake pkg-config git ca-certificates \ + libboost-dev libspeexdsp-dev zlib1g-dev \ + libssl-dev libcurl4-openssl-dev libpcre2-dev libsqlite3-dev libedit-dev libldns-dev \ + && rm -rf /var/lib/apt/lists/* + +# FreeSWITCH: headers only +RUN git clone --depth 1 --branch ${FS_VERSION} https://github.com/signalwire/freeswitch.git /src/freeswitch + +# libwebsockets: minimal build, no TLS +RUN git clone --depth 1 --branch ${LWS_VERSION} https://github.com/warmcat/libwebsockets.git /src/lws \ + && cmake -S /src/lws -B /src/lws/build \ + -DCMAKE_BUILD_TYPE=Release -DLWS_WITH_SSL=OFF \ + -DLWS_WITHOUT_TESTAPPS=ON -DLWS_WITHOUT_TEST_SERVER=ON \ + -DLWS_WITHOUT_TEST_SERVER_EXTPOLL=ON -DLWS_WITHOUT_TEST_PING=ON \ + -DLWS_WITHOUT_TEST_CLIENT=ON -DLWS_WITH_MINIMAL_EXAMPLES=OFF \ + && cmake --build /src/lws/build -j"$(nproc)" \ + && cmake --install /src/lws/build + +# Generate the build-time headers without running configure (LP64 values) +RUN cd /src/freeswitch/src/include \ + && sed -e 's/@short_value@/short/' -e 's/@int_value@/int/' -e 's/@long_value@/long/' \ + -e 's/@size_t_value@/unsigned long/' -e 's/@ssize_t_value@/long/' \ + -e 's/@voidp_size@/8/' -e 's|@prefix@|/usr/local/freeswitch|' \ + -e 's/^@[a-z0-9_]*_fmt@$//' switch_am_config.h.in > switch_am_config.h \ + && sed -E 's/@([A-Z_]+)@/0/g' switch_version.h.template > switch_version.h + +COPY mod_audio_fork /work/mod_audio_fork +WORKDIR /work/mod_audio_fork + +ENV FS_INC="-I/src/freeswitch/src/include -I/src/freeswitch/libs/libteletone/src" + +# Compile gate (-c only, no link) +RUN set -eux; \ + case "$SIMD" in sse2) SIMDFLAGS="-msse2 -DUSE_SSE2";; avx2) SIMDFLAGS="-mavx2 -DUSE_AVX2";; *) SIMDFLAGS="";; esac; \ + echo "$SIMDFLAGS" > /tmp/simdflags; \ + mkdir -p /tmp/obj; \ + gcc -c -Wall $FS_INC $(pkg-config --cflags libwebsockets) mod_audio_fork.c -o /tmp/obj/mod_audio_fork.o; \ + for f in lws_glue parser audio_pipe vector_math; do \ + g++ -c -Wall -std=c++17 $SIMDFLAGS $FS_INC $(pkg-config --cflags libwebsockets) $f.cpp -o /tmp/obj/$f.o; \ + done + +# Unit tests (link only vector_math + base64, no FreeSWITCH) +RUN set -eux; \ + g++ -Wall -std=c++17 $(cat /tmp/simdflags) -I. test/test_units.cpp vector_math.cpp -o /tmp/test_units; \ + /tmp/test_units diff --git a/README.md b/README.md index 5974d2d..b4f98b1 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,22 @@ The `mod_audio_fork` Freeswitch module, intended for use with a [jambonz](https://jambonz.org) programmable voice platform deployment. +## Development + +Tasks are run with [just](https://github.com/casey/just). Building and testing need Docker; IDE setup needs `git`, `cmake`, and (on macOS) Homebrew. + +| Target | Description | +| --- | --- | +| `just` / `just test` | Compile all sources against FreeSWITCH/libwebsockets headers and run the unit tests (native arch, scalar vector math) | +| `just test-sse2` | Same, on linux/amd64 with SSE2 vector math (emulated on arm hosts) | +| `just test-all` | Run both configurations | +| `just prereqs` | Check that Docker is installed and running | +| `just rebuild` | Rebuild from scratch, ignoring the Docker cache | +| `just try-versions ` | Build against other versions, e.g. `just try-versions v1.11.3 v5.0.0` | +| `just shell` | Interactive shell in the built image | +| `just ide-setup` | Fetch FreeSWITCH/libwebsockets headers into `.deps/` and generate `.clangd` and `.vscode/c_cpp_properties.json` for IDE support. Needs `brew install boost speexdsp` on macOS | +| `just clean` | Remove the Docker images created by the justfile | + ## Licensing This software is available under a dual-licensing scheme. For specific use in a standalone [jambonz](https://jambonz.org) deployment, the [MIT License](./LICENSE_MIT) applies. For all other uses, the software is licensed for use under the [AGPL Version 3.0 license](./LICENSE_AGPL-3.0). diff --git a/issues/review-findings.md b/issues/review-findings.md new file mode 100644 index 0000000..3554526 --- /dev/null +++ b/issues/review-findings.md @@ -0,0 +1,71 @@ +Findings from a full read-through of `mod_audio_fork/` (C/C++ sources). Not reviewed: SSE2/scalar halves of `vector_math.cpp`, `base64.hpp` body, `test/test_units.cpp`, Dockerfile, justfile, docs. + +Status is kept up to date as items are fixed. Items not touched since the original review are the review's claims and have not been re-verified against the current code. + +## Fixed on branch `robustness-fixes` + +- [x] Start with invalid URI/sample rate still starting the capture, `playAudio` NULL `audioContentType` crash, stereo+resampling buffer overflow, null derefs in error logs. +- [x] **Use-after-free on `AudioPipe`.** The pipe is owned by `shared_ptr` (session side via `PipeHandle`, pending lists, lws slot). +- [x] **Races between the lws thread and session cleanup.** Pre-buffer, set-aside byte, bidirectional resampler and playout buffer moved into a shared `StreamState` that the `AudioPipe` passes to every callback (`userData`), so cleanup only drops its reference and the lws thread never uses memory cleanup frees. The playout buffer is shared with the media thread under `StreamState::playoutMutex` (media thread only try-locks it). +- [x] **Concurrent `stop` and media-bug close.** `fork_session_cleanup` re-checks the bug under `tech_pvt->mutex`; only the first caller tears down. `tech_pvt->mutex` is no longer destroyed (session pool owns it) and is unlocked at the end of cleanup. +- [x] **`AudioPipe::context` / `stopFlag` data races.** Both atomic and documented in `audio_pipe.hpp`; `addPendingConnect` skips `lws_cancel_service` while the context does not exist yet, and the service thread wakes itself once it has created it. +- [x] **Removed `playAudio` and `transcription`** (messages, events, README). The `playAudio` temp-file list and its races went with it. `killAudio` is kept. +- [x] **Leaks / sockets left open.** `close()` on a pipe that is `IDLE` or `CONNECTING` is remembered (`m_closeRequested`) and honoured: an `IDLE` pipe never connects, a `CONNECTING` one is closed in `ESTABLISHED`. `connect_client` failing emits `CONNECT_FAIL` and drops the pipe from the pending list. `start_capture` destroys the session data when `media_bug_add` fails and runs the full cleanup when connecting fails. `fork_data_init` failure paths audited: `destroy_tech_pvt` is safe there. `CLIENT_CONNECTION_ERROR` already removed the pipe from the pending list. +- [x] **Unterminated strings.** `parse_ws_uri` no longer copies the URI into a fixed buffer, rejects hosts/paths that do not fit and copies with `switch_copy_string`; the caller's buffers are zero-initialised. `sessionId` is a session-pool string like `bugname`, and the unused `host`/`path`/`port` fields of `private_t` are gone. +- [x] **Unbounded memory growth on inbound binary.** The pipe accepts binary only if enable, stream and sample rate are all set (the condition for `SMBF_WRITE_REPLACE`); `enable` now defaults to 0. The playout buffer is deliberately left uncapped (long streamed playout is wanted). Not built or run. +- [x] **Mixing wrapped instead of clipping.** `vector_add` now saturates in the AVX2, SSE2 and scalar paths (`adds_epi16` / clamped scalar add); unit test added and passing for scalar and SSE2. The AVX2 path is not built by the test targets. +- [x] **Module unload could hang.** `deinitialize` now calls `lws_cancel_service` after setting `stopFlag` (if the context does not exist yet the service thread wakes itself and sees the flag), and the service thread resets `context` to null before `lws_context_destroy`. Not built or run. Still open: `fork_cleanup` does not close live pipes, so unloading with calls in progress is not clean. +- [x] **Unlocked `PipeHandle::get` in API calls** (`lws_glue.cpp`, from `/code-review`). `fork_session_send_text` and `fork_session_graceful_shutdown` read `tech_pvt->pAudioPipe` without `tech_pvt->mutex`, while cleanup deletes the heap `Ptr` box in `PipeHandle::release` under it, so a concurrent `stop`/hangup could leave them dereferencing a freed box (narrow window). Both now take `tech_pvt->mutex` around `get` and the call. `fork_session_pauseresume` was named in the review but never uses `PipeHandle::get`. Not built or run. +- [x] **`fork_session_cleanup` read `tech_pvt->id` before the `!tech_pvt` check.** Moved after it. Cannot happen in practice (the bug's user data is always set), but the check was dead before. +- [x] **`start` with only 4 tokens crashed FreeSWITCH.** The usage check was `argc < 4`, but `argv[4]` (sample rate) is read unconditionally, so `uuid_audio_fork start ` ran `strcmp(NULL, ...)`. Now `argc < 5`. Not run against a live FreeSWITCH. +- [x] **`start` dereferenced a NULL read codec** (`mod_audio_fork.c`). The codec was fetched before `switch_channel_pre_answer` and used unchecked; it is now fetched after it and checked. `fork_data_init` also ignored the status of `switch_core_session_get_read_impl` and used an uninitialised `read_impl` if it failed; it now returns an error (the `private_t` is zeroed first so cleanup is safe). Not run against a live FreeSWITCH. +- [x] **C++ exceptions could terminate the process.** `AudioPipe::lws_callback` is now a wrapper that catches everything from `lws_callback_impl` and closes the connection; `fork_session_init`, `parse_ws_uri` (std::regex, std::string) and `AudioPipe::enqueueText` catch and return an error. `fork_frame` and `dub_speech_frame` do not allocate and were left alone. + +## Medium + +- [x] **No upper limit on the sample rates** (`mod_audio_fork.c` start parsing). The audio buffer size is computed in `int` (`320 * rate / 8000 * channels * 1000 / 20 * secs`) and overflowed above ~6.7M (the resulting `new[]` failure is caught now, but rates of 1M to 6M still allocated tens of MB per call); `bidirectional_audio_sample_rate` was unchecked too. Both are now limited to `MAX_SAMPLE_RATE` (64000, the documented maximum) and the bidirectional rate must not be negative. Not run against a live FreeSWITCH. +- [x] **One `lws_cancel_service` and one `pendingWrites` entry per 20 ms frame per call** (`AudioPipe::unlockAudioBuffer` -> `addPendingWrite`). About 50 wakeups per second per call on the single lws thread. Checked, no performance problem seen in practice; a per-pipe "write already pending" flag would be the fix if it ever shows up. +- [ ] **First text fragment allocation is not bounded** (`audio_pipe.cpp` ~186). `malloc(len + lws_remaining_packet_payload)` uses the size the server announces; `MAX_RECV_BUF_SIZE` is only checked when the buffer grows later. After the limit is hit, the remaining fragments start a new buffer (`realloc(nullptr, ...)`) and the tail of the oversized message is delivered as a message of its own. +- [ ] **`findAndRemovePendingConnect` removes every entry with `m_wsi == nullptr`**, which includes pipes still `IDLE` and not yet connected. List order protects them in the normal case; a callback for a wsi that is not in the list would drop them, and they would never connect or report a failure. Remove only entries that are `FAILED`/`DISCONNECTED`. +- [ ] **`do_graceful_shutdown` / `addPendingWrite` / `addPendingDisconnect` dereference `m_vhd->context`**, which is null until the lws thread has started the connect (a graceful shutdown right after `start` crashes). Graceful shutdown is not used here, so low priority; use the static `context` and skip when null. +- [ ] **Graceful shutdown blocks closing.** After `graceful-shutdown` the WRITEABLE handler returns early on `isGracefulShutdown()` and never reaches the `DISCONNECTING` branch, so a later `stop` cannot close our side. Also logged at ERROR for a normal event. +- [x] **`close()` vs `LWS_CALLBACK_CLIENT_ESTABLISHED` race** (`audio_pipe.cpp` ~119, from `/code-review`). Checked, low: a second disconnect request for a connection that is already closing is harmless. The only loss is a final `stop` text that races with the connection being established. +- [x] **Short `lws_write` on a text frame closes the connection** (`audio_pipe.cpp` ~262, from `/code-review`). Checked, not an issue: lws buffers short writes and returns `n`; it returns < 0 only on a real error. +- [x] **`deinitialize` does not drain the static `pendingConnects` / `pendingDisconnects` / `pendingWrites`** (`audio_pipe.cpp` ~524, from `/code-review`). Checked, low: the static `Ptr`s are destroyed by the module's static destructors, which run before the library is unmapped; at worst they are a leak. Related to the open `fork_cleanup` item above. +- [x] **`dub_speech_frame` skips the whole frame when `clearPlayout` is set** (`lws_glue.cpp`, from `/code-review`). Checked, not an issue: after a clear there is nothing to mix, so passing the frame through untouched is correct. A try-lock miss only delays playout by one frame, because the buffered audio is kept. +- [x] **`fork_session_cleanup` returns early without `destroy_tech_pvt` when the bug private is missing** (from `/code-review`). Checked, not an issue: the early return is deliberate because whoever cleared the bug private under `tech_pvt->mutex` owns the teardown. A second caller (`stop` racing with a media-bug close) or the nested close triggered by `switch_core_media_bug_remove` finds no bug and must not destroy again. `start_capture` stores the bug right after `media_bug_add` succeeds (`set_private` cannot fail), and a failed `media_bug_add` goes through `fork_session_destroy`. The `tech_pvt->id` read before the null check is a separate minor item under Low. +- [x] **`connect_client` failure path** (`audio_pipe.cpp` ~399, from `/code-review`). Checked, low: `close()` in that window is a no-op by design (`m_closeRequested` is set), and `eventCallback` ignores events for a session whose bug is gone. +- [x] **Possible leak when `fork_data_init` fails** (`lws_glue.cpp` ~336, from `/code-review`). Checked, not an issue: `destroy_tech_pvt` releases the handles on every failure path and `start_capture` only continues after a successful init. + +## Low + +- `enqueueText` recomputes the queued byte total by iterating the deque under `m_text_mutex` on every call (up to 1024 entries); keep a running counter instead (`audio_pipe.cpp` ~583, from `/code-review`). +- `binaryWritePtrResetToZero` resets to 0 instead of `LWS_PRE`, so after an overrun the first 16 bytes are skipped; `buffer_overrun_notified` is never reset. +- `CONNECT_FAIL` JSON is concatenated by hand: message not escaped and may be NULL. Use cJSON. +- Basic auth is silently skipped if user+password exceed the 128-byte buffer, and is sent over plain `ws://`. +- `EVENT_CONNECT_SUCCESS`, `CONNECT_FAIL`, `BUFFER_OVERRUN`, `JSON` are never reserved/freed; `switch_event_create_subclass` result unchecked in `responseHandler`. +- Remaining API error logs use `session` (NULL in API context) instead of `lsession`; `parser.cpp` logs full inbound messages at ERROR. +- Metadata containing spaces is split by `switch_separate_string(' ')`, breaking the JSON. +- No `realloc` failure handling for `recv_buf` growth. +- Samples not consumed by the resampler (`in_len`) are dropped by `cBuffer->clear()`. +- Unknown message types (including the removed `playAudio` / `transcription`) are logged at ERROR. +- `EVENT_CONNECT_SUCCESS`, `CONNECT_FAIL`, `BUFFER_OVERRUN`, `JSON` are not reserved/freed in `mod_audio_fork_load`/`shutdown`; kriklivsky/freeswitch-modules `fix/code-review-bugfixes` does that (see `upstream-watch.md`). +- The kriklivsky `refactor/architecture-improvements` branch (17e5fa5) fixes the two Medium receive-buffer items (first fragment allocation, oversized tail) by using a `std::vector` with an overflow flag that discards the remaining fragments; port that idea. +- Remote `hangup` message (kriklivsky a65a94f): optional feature, not taken; see `upstream-watch.md`. + +## Compatibility notes (intentional, from `/code-review`) + +- `bidirectional_audio_enable` now defaults to 0 for the short start syntax (`argc <= 9`), and `playAudio` / `transcription` messages and their events (`play_audio`, `transcription`) were removed. Existing jambonz/FreeSWITCH callers relying on them silently lose those events. Call this out in the PR description and release notes. + +## Cleanups + +- Dead code: `nServiceThreads`, `basicAuthUser`/`basicAuthPassword`, `fork_service_threads`, `mod_audio_fork_runtime`, `!ap` after `new`, unused `sent`, and the other unused-variable warnings from the build. +- `bidirectional_audio_enable` / `bidirectional_audio_stream` in `private_t` are now written but never read. +- `base64.hpp` and its unit test are unused now. +- `dub_speech_frame` holds `playoutMutex` while mixing; copy the samples out under the lock and mix outside to shorten it. + +## Verification gaps + +- No ThreadSanitizer run and no test of the shared-state / callback path (`AudioPipe` needs FreeSWITCH + lws to build). +- No live-call run of the race fixes. +- The string-termination fix is only reviewed, not built or run (`parse_ws_uri` needs FreeSWITCH). diff --git a/issues/upstream-watch.md b/issues/upstream-watch.md new file mode 100644 index 0000000..f6d6619 --- /dev/null +++ b/issues/upstream-watch.md @@ -0,0 +1,56 @@ +# Upstream watch + +Repositories to check from time to time for features and fixes that may affect our `mod_audio_fork`. We only carry `mod_audio_fork`; ignore changes to other modules. + +Remotes in this clone: `origin` (questnet), `upstream` (lylepratt). The others are not configured by default: +`git remote add acclaim https://github.com/acclaim-ai/freeswitch-modules.git` + +How to check: `git fetch `, then for every branch (there are several, not just main) +`git log --no-merges --format='%h %ad %an %s' --date=short questnet../ -- mod_audio_fork` +and read the diffs of anything new. Update the "last checked" line below. + +## Repositories + +| Remote | URL | Notes | +|---|---|---| +| upstream | https://github.com/lylepratt/freeswitch-modules | Where our code came from. | +| acclaim | https://github.com/acclaim-ai/freeswitch-modules | Fork of RomanKok/freeswitch-modules (itself a fork). Branches: `main`, `fix-locked-thread-leak`, `fix_audio_fork`, `pre-opus`, `sip-179-ws2grpc-120ms-latency`, `standalone-makefile`, `support_opus`, `vux-63-add-silerovad-to-the-freeswitch`. | +| kriklivsky | https://github.com/kriklivsky/freeswitch-modules | Fork of upstream. `main` is old jambonz upstream (to Oct 2024), nothing new. Only `fix/code-review-bugfixes` (5 commits) and `refactor/architecture-improvements` (9, includes the first 5) touch `mod_audio_fork`; both March 2026. | +| TopCS | https://github.com/TopCS/freeswitch-modules | Fork of upstream. Branches: `main`, `dev`, `codex/update-mod_dialogflow-to-latest-google-sdk`, `feature/whisper-spool`. | + +## acclaim-ai: last checked 2026-10-08 (branch tips as of push 2026-09-30) + +Not taken yet: + +- [ ] `stop_play ` (0f0624a, `vux-63-add-silerovad…`): one line, `if (argc > 2) bugname = argv[2];`. Our `mod_audio_fork.c` still always uses the default bug name. +- [ ] Lower streaming prebuffer, 160/80 ms to 80/20 ms (07143fc, `sip-179-ws2grpc-120ms-latency`): latency vs underrun trade-off, test with jitter first. +- [ ] `fix-locked-thread-leak`: check that our connect-failure and `switch_core_media_bug_add` failure paths free the pipe (they delete the `AudioPipe` on `CONNECTION_ERROR`, on synchronous `lws_client_connect_via_info` failure, and call `fork_session_destroy()` on `bug_add` failure). The rest is already covered by our `robustness-fixes` rework. + +Deliberately skipped: raw OPUS over the WebSocket (255fb5f / 43105e1, changes the wire protocol), standalone Makefile (Docker build covers it), `mod_vad_silero`, `mod_aws_transcribe_ws`, leak-hunting log instrumentation, 2024 jambonz commits for other modules. + +## kriklivsky: last checked 2026-10-08 (branch tips as of push 2026-03-23) + +Most of it is already covered by our `robustness-fixes` work (null checks, cleanup locking, termination, exceptions, connect-failure cleanup). Not taken yet: + +- [ ] Receive buffer as `std::vector` plus an overflow flag (`refactor/architecture-improvements`, 17e5fa5 / 8ea5344): discards the remaining fragments of an oversized message. Would close the two open Medium items on the first fragment allocation and the oversized tail in `review-findings.md`. +- [ ] `{"type":"hangup"}` message hangs up the channel with NORMAL_CLEARING and fires `mod_audio_fork::hangup` (`fix/code-review-bugfixes`, a65a94f). Only new feature; lets the remote end terminate calls, so decide first whether we want it. +- [ ] Reserve and free the missing event subclasses (`CONNECT_SUCCESS`, `CONNECT_FAIL`, `BUFFER_OVERRUN`, `JSON`): matches a Low item in `review-findings.md`. + +Deliberately skipped: `std::vector` audio buffer (conflicts with our `shared_ptr` ownership), locking around `pAudioPipe = nullptr` (covered by `PipeHandle`), removal of the duplicate `else if` branches (dead code, already gone here). + +## TopCS: last checked 2026-10-08 (branch tips as of push 2026-09-30) + +Only one commit touches `mod_audio_fork` (e907b61 "build indepence", on `main`, `dev` and `feature/whisper-spool`). The rest is `mod_dialogflow` (CX API, events, packaging), a new CES module and Dialogflow backchannel work; ignore. + +Deliberately skipped: standalone CMake build with `FindSpeexDSP.cmake` plus README section (Docker build covers it; the file list is also stale, no `vector_math`). No code changes to the module. + +## Fork network sweep: checked 2026-10-08 + +Walked the whole fork network of lylepratt/freeswitch-modules with `gh api repos//forks` (48 direct forks, 8 second-level ones), compared every branch of every fork that was pushed after Feb 2024 against `kriklivsky:main` (plain jambonz upstream up to Oct 2024), and looked only at `mod_audio_fork/`. Nothing new beyond what is listed above: + +- Most other forks only carry the jambonz upstream history (about 84 commits ahead of lylepratt, no own changes to `mod_audio_fork`). They are not worth watching: sptmru (+bjzhoum, rasheed-speechlogix), Catharsis68 (+Vicfou-dev), siwting, bitvizor, hnimminh, kietcaodev, bryananderson, baron2050, ochui, callable-ltd (+sip-rtc), rammohanyadavalli, goddva, vdharashive, cyrenity, ordinerf, murugancmi, areski, ahmedggalal. The 19 forks whose last push is the fork date (2024-02-04) were not compared; they have no own commits. +- lonelyxmas (+ringover), branch `fix/azure-tts-no-device-output`: wraps the `lws_glue.cpp` resampling in try/catch (June 2024); an old jambonz commit, our `robustness-fixes` rework covers it. +- RomanKok (parent of acclaim): same branches as acclaim (`fix_audio_fork`, `support_opus`), already covered in the acclaim section. +- WorkingTheoryAS is a fork of our own repo (questnet), pushed 2026-10-03; its branch `questnet` is ours. +- justinamalonson: branches could not be listed (404), probably empty or gone. +- To re-run the sweep, compare each fork's branches against `kriklivsky:main` or against our `questnet` and filter the file list for `mod_audio_fork`. Forks with `pushed_at` older than their `created_at` + a few minutes have no own commits. diff --git a/justfile b/justfile new file mode 100644 index 0000000..abdc36c --- /dev/null +++ b/justfile @@ -0,0 +1,44 @@ +# Test and build helpers for mod_audio_fork (compile-only check + unit tests, run in Docker) + +image := "mod-audio-fork-test" +fs_version := "v1.11.3" +lws_version := "v5.0.0" + +default: test + +# Check that the required tooling is installed +prereqs: + @command -v docker >/dev/null || { echo "docker is required: https://docs.docker.com/get-docker/"; exit 1; } + @docker info >/dev/null 2>&1 || { echo "docker daemon is not running"; exit 1; } + @echo "prerequisites ok" + +# Compile all sources against FreeSWITCH/libwebsockets headers and run unit tests (native arch, scalar vector math) +test: prereqs + docker build --tag {{image}} --build-arg FS_VERSION={{fs_version}} --build-arg LWS_VERSION={{lws_version}} . + +# Same, on linux/amd64 with SSE2 vector math (emulated on arm hosts) +test-sse2: prereqs + docker build --platform linux/amd64 --tag {{image}}-sse2 --build-arg SIMD=sse2 --build-arg FS_VERSION={{fs_version}} --build-arg LWS_VERSION={{lws_version}} . + +# Run both configurations +test-all: test test-sse2 + +# Rebuild from scratch, ignoring the Docker cache +rebuild: prereqs + docker build --no-cache --tag {{image}} . + +# Try other versions, e.g. `just try-versions v1.11.3 v5.0.0` +try-versions fs lws: prereqs + docker build --tag {{image}}-try --build-arg FS_VERSION={{fs}} --build-arg LWS_VERSION={{lws}} . + +# Interactive shell in the built image for debugging +shell: test + docker run --rm -it {{image}} bash + +# Remove the images created by this file +clean: + -docker rmi {{image}} {{image}}-sse2 {{image}}-try + +# Fetch FreeSWITCH/libwebsockets headers into .deps and generate IDE config (see scripts/ide-setup.sh) +ide-setup: + scripts/ide-setup.sh {{fs_version}} {{lws_version}} diff --git a/mod_audio_fork/Makefile.am b/mod_audio_fork/Makefile.am index fc0cad5..e4fe1c0 100644 --- a/mod_audio_fork/Makefile.am +++ b/mod_audio_fork/Makefile.am @@ -4,7 +4,7 @@ MODNAME=mod_audio_fork mod_LTLIBRARIES = mod_audio_fork.la mod_audio_fork_la_SOURCES = mod_audio_fork.c lws_glue.cpp parser.cpp audio_pipe.cpp vector_math.cpp mod_audio_fork_la_CFLAGS = $(AM_CFLAGS) -mod_audio_fork_la_CXXFLAGS = $(AM_CXXFLAGS) -std=c++11 +mod_audio_fork_la_CXXFLAGS = $(AM_CXXFLAGS) -std=c++17 if USE_AVX2 mod_audio_fork_la_CXXFLAGS += -mavx2 -DUSE_AVX2 diff --git a/mod_audio_fork/README.md b/mod_audio_fork/README.md index af308a4..69c5188 100644 --- a/mod_audio_fork/README.md +++ b/mod_audio_fork/README.md @@ -24,12 +24,12 @@ Attaches media bug and starts streaming audio stream to the back-end server. Au - `sampling-rate` - choice of - "8k" = 8000 Hz sample rate will be generated - "16k" = 16000 Hz sample rate will be generated -- `metadata` - a text frame of arbitrary data to send to the back-end server immediately upon connecting. Once this text frame has been sent, the incoming audio will be sent in binary frames to the server. +- `metadata` - a text frame of arbitrary data to send to the back-end server immediately upon connecting. Once this text frame has been sent, the incoming audio will be sent in binary frames to the server. The metadata is limited to 1 MB (`MAX_TEXT_LEN`); longer metadata makes `start` fail. ``` uuid_audio_fork send_text ``` -Send a text frame of arbitrary data to the remote server (e.g. this can be used to notify of DTMF events). +Send a text frame of arbitrary data to the remote server (e.g. this can be used to notify of DTMF events). Each call is sent as exactly one websocket text frame, limited to 1 MB (`MAX_TEXT_LEN`); longer text is rejected with an error, not truncated. ``` uuid_audio_fork stop @@ -39,35 +39,6 @@ Closes websocket connection and detaches media bug, optionally sending a final t ### Events An optional feature of this module is that it can receive JSON text frames from the server and generate associated events to an application. The format of the JSON text frames and the associated events are described below. -#### audio -##### server JSON message -The server can provide audio content to be played back to the caller by sending a JSON text frame like this: -```json -{ - "type": "playAudio", - "data": { - "audioContentType": "raw", - "sampleRate": 8000, - "audioContent": "base64 encoded raw audio..", - "textContent": "Hi there! How can we help?" - } -} -``` -The `audioContentType` value can be either `wave` or `raw`. If the latter, then `sampleRate` must be specified. The audio content itself is supplied as a base64 encoded string. The `textContent` attribute can optionally contain the text of the prompt. This allows an application to choose whether to play the raw audio or to use its own text-to-speech to play the text prompt. - -Note that the module does _not_ directly play out the raw audio. Instead, it writes it to a temporary file and provides the path to the file in the event generated. It is left to the application to play out this file if it wishes to do so. -##### Freeswitch event generated -**Name**: mod_audio_fork::play_audio -**Body**: JSON string -``` -{ - "audioContentType": "raw", - "sampleRate": 8000, - "textContent": "Hi there! How can we help?", - "file": "/tmp/7dd5e34e-5db4-4edb-a166-757e5d29b941_2.tmp.r8" -} -``` -Note the audioContent attribute has been replaced with the path to the file containing the audio. This temporary file will be removed when the Freeswitch session ends. #### killAudio ##### server JSON message The server can provide a request to kill the current audio playback: @@ -83,23 +54,6 @@ Any current audio being played to the caller will be immediately stopped. The e **Body**: JSON string - the data attribute from the server message -#### transcription -##### server JSON message -The server can optionally provide transcriptions to the application in real-time: -```json -{ - "type": "transcription", - "data": { - - } -} -``` -The transcription data can be any JSON object; for instance, a server may choose to return a transcript and an associated confidence level. Whatever is provided as the `data` attribute will be attached to the generated event. - -##### Freeswitch event generated -**Name**: mod_audio_fork::transcription -**Body**: JSON string - the data attribute from the server message - #### transfer ##### server JSON message The server can optionally provide a request to transfer the call: diff --git a/mod_audio_fork/audio_pipe.cpp b/mod_audio_fork/audio_pipe.cpp index b014bb4..b442fca 100644 --- a/mod_audio_fork/audio_pipe.cpp +++ b/mod_audio_fork/audio_pipe.cpp @@ -1,9 +1,15 @@ #include "audio_pipe.hpp" +#include "mod_audio_fork.h" #include +#include #include #include +/* upper bounds for text waiting to be sent on one pipe */ +#define MAX_QUEUED_TEXT_BYTES (4 * 1024 * 1024) +#define MAX_QUEUED_TEXT_MESSAGES 1024 + /* discard incoming text messages over the socket that are longer than this */ #define MAX_RECV_BUF_SIZE (65 * 1024 * 10) #define RECV_BUF_REALLOC_SIZE (8 * 1024) @@ -41,12 +47,26 @@ static int dch_lws_http_basic_auth_gen(const char *user, const char *pw, char *b int AudioPipe::lws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len) { + try { + return lws_callback_impl(wsi, reason, user, in, len); + } catch (const std::exception& e) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_callback reason %d: exception %s, closing connection\n", (int) reason, e.what()); + } catch (...) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_callback reason %d: exception, closing connection\n", (int) reason); + } + return -1; +} + +int AudioPipe::lws_callback_impl(struct lws *wsi, + enum lws_callback_reasons reason, + void *user, void *in, size_t len) { struct AudioPipe::lws_per_vhost_data *vhd = (struct AudioPipe::lws_per_vhost_data *) lws_protocol_vh_priv_get(lws_get_vhost(wsi), lws_get_protocol(wsi)); struct lws_vhost* vhost = lws_get_vhost(wsi); - AudioPipe ** ppAp = (AudioPipe **) user; + // the slot holds a heap-allocated shared_ptr, set on establish and released on close + Ptr ** ppAp = (Ptr **) user; switch (reason) { case LWS_CALLBACK_PROTOCOL_INIT: @@ -58,7 +78,7 @@ int AudioPipe::lws_callback(struct lws *wsi, case LWS_CALLBACK_CLIENT_APPEND_HANDSHAKE_HEADER: { - AudioPipe* ap = findPendingConnect(wsi); + Ptr ap = findPendingConnect(wsi); if (ap && ap->hasBasicAuth()) { unsigned char **p = (unsigned char **)in, *end = (*p) + len; char b[128]; @@ -83,12 +103,12 @@ int AudioPipe::lws_callback(struct lws *wsi, break; case LWS_CALLBACK_CLIENT_CONNECTION_ERROR: { - AudioPipe* ap = findAndRemovePendingConnect(wsi); + Ptr ap = findAndRemovePendingConnect(wsi); int rc = lws_http_client_http_response(wsi); switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CONNECTION_ERROR: %s, response status %d\n", in ? (char *)in : "(null)", rc); if (ap) { ap->m_state = LWS_CLIENT_FAILED; - ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECT_FAIL, (char *) in, NULL, len); + ap->notify(AudioPipe::CONNECT_FAIL, (char *) in, NULL, len); } else { switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CONNECTION_ERROR unable to find wsi %p..\n", wsi); @@ -98,50 +118,62 @@ int AudioPipe::lws_callback(struct lws *wsi, case LWS_CALLBACK_CLIENT_ESTABLISHED: { - AudioPipe* ap = findAndRemovePendingConnect(wsi); + Ptr ap = findAndRemovePendingConnect(wsi); if (ap) { - *ppAp = ap; + *ppAp = new Ptr(ap); ap->m_vhd = vhd; - ap->m_state = LWS_CLIENT_CONNECTED; - ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECT_SUCCESS, NULL, NULL, len); + { + // m_state changes under the text lock, see enqueueText + std::lock_guard lk(ap->m_text_mutex); + ap->m_state = LWS_CLIENT_CONNECTED; + if (!ap->m_text_queue.empty()) lws_callback_on_writable(wsi); + } + // close() was called while we were connecting and found nothing to close: do it now. Returning -1 + // closes the connection, LWS_CALLBACK_CLIENT_CLOSED follows. Either this sees the flag or close() + // sees CONNECTED and queues a disconnect, so the connection never stays open unnoticed. + if (ap->m_closeRequested) { + ap->m_state = LWS_CLIENT_DISCONNECTING; + return -1; + } + ap->notify(AudioPipe::CONNECT_SUCCESS, NULL, NULL, len); } else { - switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_ESTABLISHED %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_ESTABLISHED unable to find wsi %p..\n", wsi); } } break; case LWS_CALLBACK_CLIENT_CLOSED: { - AudioPipe* ap = *ppAp; - if (!ap) { - switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CLOSED %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); + if (!*ppAp) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_CLOSED unable to find wsi %p..\n", wsi); return 0; } + Ptr ap = **ppAp; if (ap->m_state == LWS_CLIENT_DISCONNECTING) { // closed by us - ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECTION_CLOSED_GRACEFULLY, NULL, NULL, len); + ap->notify(AudioPipe::CONNECTION_CLOSED_GRACEFULLY, NULL, NULL, len); } else if (ap->m_state == LWS_CLIENT_CONNECTED) { // closed by far end switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,"%s socket closed by far end\n", ap->m_uuid.c_str()); - ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::CONNECTION_DROPPED, NULL, NULL, len); + ap->notify(AudioPipe::CONNECTION_DROPPED, NULL, NULL, len); } ap->m_state = LWS_CLIENT_DISCONNECTED; + ap->m_wsi = nullptr; - //NB: after receiving any of the events above, any holder of a - //pointer or reference to this object must treat is as no longer valid - + // NB: after receiving any of the events above, the session side must treat the connection as + // gone (the state is DISCONNECTED). The object itself stays valid until the last reference is released. + delete *ppAp; *ppAp = NULL; - delete ap; } break; case LWS_CALLBACK_CLIENT_RECEIVE: { - AudioPipe* ap = *ppAp; + AudioPipe* ap = *ppAp ? (*ppAp)->get() : nullptr; if (!ap) { - switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE unable to find wsi %p..\n", wsi); return 0; } @@ -152,7 +184,7 @@ int AudioPipe::lws_callback(struct lws *wsi, if (lws_frame_is_binary(wsi)) { if (len > 0 && ap->is_bidirectional_audio_stream()) { - ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::BINARY, NULL, (char *) in, len); + ap->notify(AudioPipe::BINARY, NULL, (char *) in, len); } else if (len > 0) { switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_RECEIVE received unexpected binary frame, discarding.\n"); } @@ -197,7 +229,7 @@ int AudioPipe::lws_callback(struct lws *wsi, if (lws_is_final_fragment(wsi)) { if (nullptr != ap->m_recv_buf) { std::string msg((char *)ap->m_recv_buf, ap->m_recv_buf_ptr - ap->m_recv_buf); - ap->m_callback(ap->m_uuid.c_str(), ap->m_bugname.c_str(), AudioPipe::MESSAGE, msg.c_str(), NULL, len); + ap->notify(AudioPipe::MESSAGE, msg.c_str(), NULL, len); if (nullptr != ap->m_recv_buf) free(ap->m_recv_buf); } ap->m_recv_buf = ap->m_recv_buf_ptr = nullptr; @@ -210,9 +242,9 @@ int AudioPipe::lws_callback(struct lws *wsi, case LWS_CALLBACK_CLIENT_WRITEABLE: { - AudioPipe* ap = *ppAp; + AudioPipe* ap = *ppAp ? (*ppAp)->get() : nullptr; if (!ap) { - switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_WRITEABLE %s unable to find wsi %p..\n", ap->m_uuid.c_str(), wsi); + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread LWS_CALLBACK_CLIENT_WRITEABLE unable to find wsi %p..\n", wsi); return 0; } @@ -227,12 +259,16 @@ int AudioPipe::lws_callback(struct lws *wsi, // check for text frames to send { std::lock_guard lk(ap->m_text_mutex); - if (ap->m_metadata.length() > 0) { - uint8_t buf[ap->m_metadata.length() + LWS_PRE]; - memcpy(buf + LWS_PRE, ap->m_metadata.c_str(), ap->m_metadata.length()); - int n = ap->m_metadata.length(); - int m = lws_write(wsi, buf + LWS_PRE, n, LWS_WRITE_TEXT); - ap->m_metadata.clear(); + if (!ap->m_text_queue.empty()) { + // One queued message per websocket frame: send_text calls must never be merged. + // The buffer is on the heap (not a stack array) because messages can be up to MAX_TEXT_LEN. + // A short lws_write is buffered by lws itself, so one write per writeable event is enough. + std::string text = std::move(ap->m_text_queue.front()); + ap->m_text_queue.pop_front(); + std::vector buf(LWS_PRE + text.length()); + memcpy(buf.data() + LWS_PRE, text.data(), text.length()); + int n = text.length(); + int m = lws_write(wsi, buf.data() + LWS_PRE, n, LWS_WRITE_TEXT); if (m < n) { return -1; } @@ -285,38 +321,46 @@ static const lws_retry_bo_t retry = { 0 // jitter_percent }; -struct lws_context *AudioPipe::context = nullptr; +std::atomic AudioPipe::context{nullptr}; std::thread AudioPipe::serviceThread; std::string AudioPipe::protocolName; std::mutex AudioPipe::mutex_connects; std::mutex AudioPipe::mutex_disconnects; std::mutex AudioPipe::mutex_writes; -std::list AudioPipe::pendingConnects; -std::list AudioPipe::pendingDisconnects; -std::list AudioPipe::pendingWrites; +std::list AudioPipe::pendingConnects; +std::list AudioPipe::pendingDisconnects; +std::list AudioPipe::pendingWrites; AudioPipe::log_emit_function AudioPipe::logger; -std::mutex AudioPipe::mapMutex; -bool AudioPipe::stopFlag; +std::atomic AudioPipe::stopFlag{false}; void AudioPipe::processPendingConnects(lws_per_vhost_data *vhd) { - std::list connects; + std::list connects; { std::lock_guard guard(mutex_connects); for (auto it = pendingConnects.begin(); it != pendingConnects.end(); ++it) { - if ((*it)->m_state == LWS_CLIENT_IDLE) { - connects.push_back(*it); - (*it)->m_state = LWS_CLIENT_CONNECTING; - } + // atomic, so a close() that wins the race (IDLE -> DISCONNECTED) is never connected + LwsState_t expected = LWS_CLIENT_IDLE; + if ((*it)->m_state.compare_exchange_strong(expected, LWS_CLIENT_CONNECTING)) connects.push_back(*it); } + // drop pipes that were closed before they got to connect + pendingConnects.remove_if([](const Ptr& p) { return p->m_state == LWS_CLIENT_DISCONNECTED; }); } for (auto it = connects.begin(); it != connects.end(); ++it) { - AudioPipe* ap = *it; - ap->connect_client(vhd); + Ptr ap = *it; + if (!ap->connect_client(vhd)) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"%s unable to start connection\n", ap->m_uuid.c_str()); + { + std::lock_guard guard(mutex_connects); + pendingConnects.remove(ap); + } + ap->m_state = LWS_CLIENT_FAILED; + ap->notify(AudioPipe::CONNECT_FAIL, "unable to start connection", NULL, 0); + } } } void AudioPipe::processPendingDisconnects(lws_per_vhost_data *vhd) { - std::list disconnects; + std::list disconnects; { std::lock_guard guard(mutex_disconnects); for (auto it = pendingDisconnects.begin(); it != pendingDisconnects.end(); ++it) { @@ -325,13 +369,13 @@ void AudioPipe::processPendingDisconnects(lws_per_vhost_data *vhd) { pendingDisconnects.clear(); } for (auto it = disconnects.begin(); it != disconnects.end(); ++it) { - AudioPipe* ap = *it; - lws_callback_on_writable(ap->m_wsi); + Ptr ap = *it; + if (ap->m_wsi) lws_callback_on_writable(ap->m_wsi); } } void AudioPipe::processPendingWrites() { - std::list writes; + std::list writes; { std::lock_guard guard(mutex_writes); for (auto it = pendingWrites.begin(); it != pendingWrites.end(); ++it) { @@ -340,15 +384,15 @@ void AudioPipe::processPendingWrites() { pendingWrites.clear(); } for (auto it = writes.begin(); it != writes.end(); ++it) { - AudioPipe* ap = *it; - lws_callback_on_writable(ap->m_wsi); + Ptr ap = *it; + if (ap->m_wsi) lws_callback_on_writable(ap->m_wsi); } } -AudioPipe* AudioPipe::findAndRemovePendingConnect(struct lws *wsi) { - AudioPipe* ap = NULL; +AudioPipe::Ptr AudioPipe::findAndRemovePendingConnect(struct lws *wsi) { + Ptr ap; std::lock_guard guard(mutex_connects); - std::list toRemove; + std::list toRemove; for (auto it = pendingConnects.begin(); it != pendingConnects.end() && !ap; ++it) { int state = (*it)->m_state; @@ -370,8 +414,8 @@ AudioPipe* AudioPipe::findAndRemovePendingConnect(struct lws *wsi) { return ap; } -AudioPipe* AudioPipe::findPendingConnect(struct lws *wsi) { - AudioPipe* ap = NULL; +AudioPipe::Ptr AudioPipe::findPendingConnect(struct lws *wsi) { + Ptr ap; std::lock_guard guard(mutex_connects); for (auto it = pendingConnects.begin(); it != pendingConnects.end() && !ap; ++it) { @@ -382,17 +426,18 @@ AudioPipe* AudioPipe::findPendingConnect(struct lws *wsi) { return ap; } -void AudioPipe::addPendingConnect(AudioPipe* ap) { +void AudioPipe::addPendingConnect(Ptr ap) { { std::lock_guard guard(mutex_connects); pendingConnects.push_back(ap); switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG,"%s after adding connect there are %lu pending connects\n", ap->m_uuid.c_str(), pendingConnects.size()); } - lws_cancel_service(context); + // the service thread has not created the context yet if this is the very first call after module load; + // the connect is queued already and the service thread wakes itself once the context exists + if (struct lws_context* ctx = context.load()) lws_cancel_service(ctx); } -void AudioPipe::addPendingDisconnect(AudioPipe* ap) { - ap->m_state = LWS_CLIENT_DISCONNECTING; +void AudioPipe::addPendingDisconnect(Ptr ap) { { std::lock_guard guard(mutex_disconnects); pendingDisconnects.push_back(ap); @@ -401,7 +446,7 @@ void AudioPipe::addPendingDisconnect(AudioPipe* ap) { } lws_cancel_service(ap->m_vhd->context); } -void AudioPipe::addPendingWrite(AudioPipe* ap) { +void AudioPipe::addPendingWrite(Ptr ap) { { std::lock_guard guard(mutex_writes); pendingWrites.push_back(ap); @@ -437,19 +482,25 @@ bool AudioPipe::lws_service_thread() { switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,"AudioPipe::lws_service_thread creating context\n"); - context = lws_create_context(&info); - if (!context) { + struct lws_context* ctx = lws_create_context(&info); + context = ctx; + if (!ctx) { switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR,"AudioPipe::lws_service_thread failed creating context\n"); return false; } + // connects queued before the context existed could not wake the service loop + lws_cancel_service(ctx); + int n; do { - n = lws_service(context, 0); + n = lws_service(ctx, 0); } while (n >= 0 && !stopFlag); lwsl_notice("AudioPipe::lws_service_thread ending\n"); - lws_context_destroy(context); + // addPendingConnect() must not wake a context that is about to be destroyed + context = nullptr; + lws_context_destroy(ctx); return true; } @@ -459,15 +510,16 @@ void AudioPipe::initialize(const char* protocol, int loglevel, log_emit_function lws_set_log_level(loglevel, logger); switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE,"AudioPipe::initialize starting\n"); - std::lock_guard lock(mapMutex); stopFlag = false; serviceThread = std::thread(&AudioPipe::lws_service_thread); } bool AudioPipe::deinitialize() { switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE,"AudioPipe::deinitialize\n"); - std::lock_guard lock(mapMutex); stopFlag = true; + // lws_service(ctx, 0) blocks until an lws event; wake it so it sees stopFlag. If the context does not exist + // yet, the service thread wakes itself once it has created it and then sees stopFlag. + if (struct lws_context* ctx = context.load()) lws_cancel_service(ctx); if (serviceThread.joinable()) { serviceThread.join(); } @@ -477,11 +529,11 @@ bool AudioPipe::deinitialize() { // instance members AudioPipe::AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path, int sslFlags, size_t bufLen, size_t minFreespace, const char* username, const char* password, char* bugname, - int bidirectional_audio_stream, notifyHandler_t callback) : + int bidirectional_audio_stream, notifyHandler_t callback, std::shared_ptr userData) : m_uuid(uuid), m_host(host), m_port(port), m_path(path), m_sslFlags(sslFlags), m_audio_buffer_min_freespace(minFreespace), m_audio_buffer_max_len(bufLen), m_gracefulShutdown(false), m_audio_buffer_write_offset(LWS_PRE), m_recv_buf(nullptr), m_recv_buf_ptr(nullptr), m_bugname(bugname), - m_state(LWS_CLIENT_IDLE), m_wsi(nullptr), m_vhd(nullptr), m_callback(callback) { + m_state(LWS_CLIENT_IDLE), m_wsi(nullptr), m_vhd(nullptr), m_callback(callback), m_userData(std::move(userData)) { if (username && password) { m_username.assign(username); @@ -496,7 +548,7 @@ AudioPipe::~AudioPipe() { } void AudioPipe::connect(void) { - addPendingConnect(this); + addPendingConnect(shared_from_this()); } bool AudioPipe::connect_client(struct lws_per_vhost_data *vhd) { @@ -525,26 +577,51 @@ bool AudioPipe::connect_client(struct lws_per_vhost_data *vhd) { return nullptr != m_wsi; } -void AudioPipe::bufferForSending(const char* text) { - if (m_state != LWS_CLIENT_CONNECTED) return; +bool AudioPipe::enqueueText(const char* text) { + try { + return enqueueTextImpl(text); + } catch (...) { + return false; // out of memory + } +} + +bool AudioPipe::enqueueTextImpl(const char* text) { + size_t len = strlen(text); + if (len > MAX_TEXT_LEN) return false; + bool connected; { std::lock_guard lk(m_text_mutex); - m_metadata.append(text); + // m_state changes to CONNECTED under this lock, so a text queued before that is flushed by the + // connect handler and a text queued after it needs the pending write below + connected = m_state == LWS_CLIENT_CONNECTED; + if (!connected && m_state != LWS_CLIENT_IDLE && m_state != LWS_CLIENT_CONNECTING) return false; + if (m_text_queue.size() >= MAX_QUEUED_TEXT_MESSAGES) return false; + size_t queued = 0; + for (const auto& queuedText : m_text_queue) queued += queuedText.length(); + if (queued + len > MAX_QUEUED_TEXT_BYTES) return false; + m_text_queue.emplace_back(text); } - addPendingWrite(this); + if (connected) addPendingWrite(shared_from_this()); + return true; } void AudioPipe::unlockAudioBuffer() { - if (m_audio_buffer_write_offset > LWS_PRE) addPendingWrite(this); + if (m_audio_buffer_write_offset > LWS_PRE) addPendingWrite(shared_from_this()); m_audio_mutex.unlock(); } void AudioPipe::close() { - if (m_state != LWS_CLIENT_CONNECTED) return; - addPendingDisconnect(this); + // set first: a connection that is still being established is closed by the lws thread once it is up + m_closeRequested = true; + // atomic, so a connection closed concurrently by the lws thread is never moved back to DISCONNECTING + LwsState_t expected = LWS_CLIENT_IDLE; + if (m_state.compare_exchange_strong(expected, LWS_CLIENT_DISCONNECTED)) return; // never connected + expected = LWS_CLIENT_CONNECTED; + if (!m_state.compare_exchange_strong(expected, LWS_CLIENT_DISCONNECTING)) return; + addPendingDisconnect(shared_from_this()); } void AudioPipe::do_graceful_shutdown() { m_gracefulShutdown = true; - addPendingWrite(this); + addPendingWrite(shared_from_this()); } diff --git a/mod_audio_fork/audio_pipe.hpp b/mod_audio_fork/audio_pipe.hpp index bfeab49..fd90ce1 100644 --- a/mod_audio_fork/audio_pipe.hpp +++ b/mod_audio_fork/audio_pipe.hpp @@ -3,17 +3,22 @@ #include #include +#include +#include #include #include #include #include +#include #include namespace drachtio { - class AudioPipe { + class AudioPipe : public std::enable_shared_from_this { public: + using Ptr = std::shared_ptr; + enum LwsState_t { LWS_CLIENT_IDLE, LWS_CLIENT_CONNECTING, @@ -31,7 +36,10 @@ namespace drachtio { BINARY }; typedef void (*log_emit_function)(int level, const char *line); - typedef void (*notifyHandler_t)(const char *sessionId, const char* bugname, NotifyEvent_t event, const char* message, const char* binary, size_t binary_len ); + // userData is the object the session handed to the constructor. It is only valid for the duration of + // the call, but stays alive at least that long even if the session has already let go of it, so + // it is the safe way for the lws thread to reach state that the session side may be tearing down + typedef void (*notifyHandler_t)(const char *sessionId, const char* bugname, NotifyEvent_t event, const char* message, const char* binary, size_t binary_len, void* userData); struct lws_per_vhost_data { struct lws_context *context; @@ -43,15 +51,17 @@ namespace drachtio { static bool deinitialize(); static bool lws_service_thread(); - // constructor + // constructor: instances must be owned by a shared_ptr (std::make_shared) AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path, int sslFlags, size_t bufLen, size_t minFreespace, const char* username, const char* password, char* bugname, - int bidirectional_audio, notifyHandler_t callback); + int bidirectional_audio, notifyHandler_t callback, std::shared_ptr userData); ~AudioPipe(); LwsState_t getLwsState(void) { return m_state; } void connect(void); - void bufferForSending(const char* text); + // queues text as one websocket text frame (held until connected), returns false if it exceeds + // MAX_TEXT_LEN, the queue is full (message count or total bytes), or the pipe is no longer usable + bool enqueueText(const char* text); size_t binarySpaceAvailable(void) { return m_audio_buffer_max_len - m_audio_buffer_write_offset; } @@ -99,38 +109,55 @@ namespace drachtio { private: static std::thread serviceThread; + // lws calls this from C, so no exception may leave it: it wraps lws_callback_impl and closes the connection + // (returns -1) if that throws, e.g. when out of memory static int lws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len); - static struct lws_context *context; + static int lws_callback_impl(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len); + // The lws context is created by the service thread (lws_service_thread), but addPendingConnect() + // reads it from whichever thread starts a call. Atomic so that read is not a data race. + // It is null until the service thread has created the context (a call started right after module load + // can see null), and it is reset to null by the service thread before the context is destroyed on unload. + static std::atomic context; static std::string protocolName; static std::mutex mutex_connects; static std::mutex mutex_disconnects; static std::mutex mutex_writes; - static std::list pendingConnects; - static std::list pendingDisconnects; - static std::list pendingWrites; + // the pending lists and the lws per-connection slot own references, so a pipe stays alive until + // every queued entry has been processed, regardless of when the session side lets go of it + static std::list pendingConnects; + static std::list pendingDisconnects; + static std::list pendingWrites; static log_emit_function logger; - static std::mutex mapMutex; - static bool stopFlag; - - static AudioPipe* findAndRemovePendingConnect(struct lws *wsi); - static AudioPipe* findPendingConnect(struct lws *wsi); - static void addPendingConnect(AudioPipe* ap); - static void addPendingDisconnect(AudioPipe* ap); - static void addPendingWrite(AudioPipe* ap); + // Module-wide, not per call: asks the lws service thread to exit its loop. Cleared by initialize() + // (module load), set by deinitialize() (module unload), and checked by lws_service_thread() each time + // lws_service() returns. Setting it does not wake lws_service(), so deinitialize() also calls lws_cancel_service(). + // Atomic because it is written by the unloading thread and read by the service thread. + static std::atomic stopFlag; + + static Ptr findAndRemovePendingConnect(struct lws *wsi); + static Ptr findPendingConnect(struct lws *wsi); + static void addPendingConnect(Ptr ap); + static void addPendingDisconnect(Ptr ap); + static void addPendingWrite(Ptr ap); static void processPendingConnects(lws_per_vhost_data *vhd); static void processPendingDisconnects(lws_per_vhost_data *vhd); static void processPendingWrites(void); + bool enqueueTextImpl(const char* text); bool connect_client(struct lws_per_vhost_data *vhd); + void notify(NotifyEvent_t event, const char* message, const char* binary, size_t len) { + m_callback(m_uuid.c_str(), m_bugname.c_str(), event, message, binary, len, m_userData.get()); + } - LwsState_t m_state; + std::atomic m_state; + std::atomic m_closeRequested{false}; std::string m_uuid; std::string m_host; std::string m_bugname; unsigned int m_port; std::string m_path; - std::string m_metadata; + std::deque m_text_queue; std::mutex m_text_mutex; std::mutex m_audio_mutex; int m_sslFlags; @@ -144,6 +171,8 @@ namespace drachtio { size_t m_recv_buf_len; struct lws_per_vhost_data* m_vhd; notifyHandler_t m_callback; + // shared with the session; released with the pipe, i.e. after the last queued entry and lws callback is done + std::shared_ptr m_userData; log_emit_function m_logger; std::string m_username; std::string m_password; diff --git a/mod_audio_fork/lws_glue.cpp b/mod_audio_fork/lws_glue.cpp index 4dbac06..9c5b932 100644 --- a/mod_audio_fork/lws_glue.cpp +++ b/mod_audio_fork/lws_glue.cpp @@ -3,17 +3,18 @@ #include #include #include +#include +#include #include #include +#include #include #include #include #include -#include #include #include -#include "base64.hpp" #include "parser.hpp" #include "mod_audio_fork.h" #include "audio_pipe.hpp" @@ -36,15 +37,71 @@ namespace { std::getenv("MOD_AUDIO_FORK_SUBPROTOCOL_NAME") : "audio.drachtio.org"; static unsigned int nServiceThreads = std::max(1, std::min(requestedNumServiceThreads ? ::atoi(requestedNumServiceThreads) : 1, 5)); static unsigned int idxCallCount = 0; - static uint32_t playCount = 0; - switch_status_t processIncomingBinary(private_t* tech_pvt, switch_core_session_t* session, const char* message, size_t dataLength) { + // Session-side owner of an AudioPipe. private_t is a plain C struct shared with mod_audio_fork.c, + // so the reference lives behind its void* pAudioPipe. It is held until destroy_tech_pvt; a pipe whose + // connection has ended is detected by its state, not by a null pointer. + namespace PipeHandle { + typedef std::shared_ptr Ptr; + + void set(private_t* tech_pvt, Ptr pipe) { tech_pvt->pAudioPipe = new Ptr(std::move(pipe)); } + + drachtio::AudioPipe* get(const private_t* tech_pvt) { + return tech_pvt->pAudioPipe ? static_cast(tech_pvt->pAudioPipe)->get() : nullptr; + } + + void release(private_t* tech_pvt) { + delete static_cast(tech_pvt->pAudioPipe); + tech_pvt->pAudioPipe = nullptr; + } + } + + // Everything the lws thread touches for bidirectional audio. It is shared (shared_ptr) between the session + // (tech_pvt->pStream) and the AudioPipe, which passes it to every callback. So cleanup on the session side + // only drops its reference and the state stays alive until the pipe is finished with it, i.e. it is freed + // by whoever lets go last instead of by fork_session_cleanup while the lws thread may still be using it. + struct StreamState { + // lws thread only + CircularBuffer_t preBuffer{8192}; + SpeexResamplerState* resampler = nullptr; + uint8_t setAsideByte = 0; + bool hasSetAsideByte = false; + int preBufSize = 0; + int downscaleFactor = 1; + + // shared between the lws thread (insert) and the media thread (drain) + std::mutex playoutMutex; + CircularBuffer_t playout{8192}; // guarded by playoutMutex + std::atomic clearPlayout{false}; + + ~StreamState() { + if (resampler) speex_resampler_destroy(resampler); + } + }; + + // session-side owner of a StreamState reference, see PipeHandle + namespace StreamHandle { + typedef std::shared_ptr Ptr; + + void set(private_t* tech_pvt, Ptr stream) { tech_pvt->pStream = new Ptr(std::move(stream)); } + + StreamState* get(const private_t* tech_pvt) { + return tech_pvt->pStream ? static_cast(tech_pvt->pStream)->get() : nullptr; + } + + void release(private_t* tech_pvt) { + delete static_cast(tech_pvt->pStream); + tech_pvt->pStream = nullptr; + } + } + + switch_status_t processIncomingBinary(StreamState* ss, const char* message, size_t dataLength) { std::vector data; // Prepend the set-aside byte if there is one - if (tech_pvt->has_set_aside_byte) { - data.push_back(tech_pvt->set_aside_byte); - tech_pvt->has_set_aside_byte = false; + if (ss->hasSetAsideByte) { + data.push_back(ss->setAsideByte); + ss->hasSetAsideByte = false; } // Append the new incoming message @@ -53,8 +110,8 @@ namespace { // Check if the total data length is now odd if (data.size() % 2 != 0) { // Set aside the last byte - tech_pvt->set_aside_byte = data.back(); - tech_pvt->has_set_aside_byte = true; + ss->setAsideByte = data.back(); + ss->hasSetAsideByte = true; data.pop_back(); // Remove the last byte from the data vector } @@ -63,7 +120,7 @@ namespace { size_t numSamples = data.size() / sizeof(uint16_t); // Access the prebuffer - CircularBuffer_t* cBuffer = static_cast(tech_pvt->streamingPreBuffer); + CircularBuffer_t* cBuffer = &ss->preBuffer; // Ensure the prebuffer has enough capacity if (cBuffer->capacity() - cBuffer->size() < numSamples) { @@ -76,7 +133,7 @@ namespace { //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Appended %zu 16-bit samples to the prebuffer.\n", numSamples); // if we haven't reached threshold amount of prebuffered data, return - if (cBuffer->size() < tech_pvt->streamingPreBufSize) { + if (cBuffer->size() < ss->preBufSize) { //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Prebuffered data is below threshold %u, returning.\n", tech_pvt->streamingPreBufSize); return SWITCH_STATUS_SUCCESS; } @@ -84,10 +141,10 @@ namespace { //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Prebuffered data samples %u is above threshold %u, prepare to playout.\n", cBuffer->size(), tech_pvt->streamingPreBufSize); // after initial pre-buffering, rachet down the threshold to 40ms - tech_pvt->streamingPreBufSize = 320 * tech_pvt->downscale_factor * 2; + ss->preBufSize = 320 * ss->downscaleFactor * 2; // Check for downsampling factor - size_t downsample_factor = tech_pvt->downscale_factor; + size_t downsample_factor = ss->downscaleFactor; // Calculate the number of samples that can be evenly divided by the downsample factor size_t numCompleteSamples = (cBuffer->size() / downsample_factor) * downsample_factor; @@ -104,7 +161,7 @@ namespace { // resample if necessary std::vector out; try { - if (tech_pvt->bidirectional_audio_resampler) { + if (ss->resampler) { // Improvement: Use assign to convert circular buffer to vector for resampling std::vector in; in.assign(cBuffer->begin(), cBuffer->end()); @@ -115,7 +172,7 @@ namespace { //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Resampling %u samples into a buffer that can hold %u samples\n", in.size(), out_len); - speex_resampler_process_interleaved_int(tech_pvt->bidirectional_audio_resampler, in.data(), &in_len, out.data(), &out_len); + speex_resampler_process_interleaved_int(ss->resampler, in.data(), &in_len, out.data(), &out_len); // Resize the output buffer to match the output length from resampler //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Resizing output buffer from %u to %u samples\n", in.size(), out_len); @@ -134,8 +191,10 @@ namespace { return SWITCH_STATUS_FALSE; } - if (nullptr != tech_pvt->mutex && switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) { - CircularBuffer_t *playoutBuffer = (CircularBuffer_t *) tech_pvt->streamingPlayoutBuffer; + { + // only held for the insert; the media thread holds it just as briefly while draining + std::lock_guard lk(ss->playoutMutex); + CircularBuffer_t *playoutBuffer = &ss->playout; try { // Resize the buffer if necessary @@ -148,150 +207,60 @@ namespace { playoutBuffer->insert(playoutBuffer->end(), out.begin(), out.end()); //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Appended %zu 16-bit samples to the playout buffer.\n", out.size()); } catch (const std::exception& e) { - switch_mutex_unlock(tech_pvt->mutex); cBuffer->clear(); switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error processing incoming binary message: %s\n", e.what()); return SWITCH_STATUS_FALSE; } catch (...) { - switch_mutex_unlock(tech_pvt->mutex); cBuffer->clear(); switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error processing incoming binary message\n"); return SWITCH_STATUS_FALSE; } - switch_mutex_unlock(tech_pvt->mutex); - cBuffer->clear(); - - // Put the leftover samples back in the prebuffer for the next time - if (!leftoverSamples.empty()) { - cBuffer->insert(cBuffer->end(), leftoverSamples.begin(), leftoverSamples.end()); - //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Put back %u leftover samples into the prebuffer.\n", leftoverSamples.size()); - } - return SWITCH_STATUS_SUCCESS; } - switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Failed to get mutext (temp)\n"); + cBuffer->clear(); + // Put the leftover samples back in the prebuffer for the next time + if (!leftoverSamples.empty()) { + cBuffer->insert(cBuffer->end(), leftoverSamples.begin(), leftoverSamples.end()); + //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Put back %u leftover samples into the prebuffer.\n", leftoverSamples.size()); + } return SWITCH_STATUS_SUCCESS; } - void processIncomingMessage(private_t* tech_pvt, switch_core_session_t* session, const char* message) { + void processIncomingMessage(private_t* tech_pvt, switch_core_session_t* session, const char* message, StreamState* ss) { std::string msg = message; std::string type; cJSON* json = parse_json(session, msg, type) ; if (json) { switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) processIncomingMessage - received %s message %s\n", tech_pvt->id, type.c_str(), message); cJSON* jsonData = cJSON_GetObjectItem(json, "data"); - if (0 == type.compare("playAudio") && - // playAudio is enabled and there is no bidirectional audio from stream is enabled. - tech_pvt->bidirectional_audio_enable && - !tech_pvt->bidirectional_audio_stream) { - if (jsonData) { - // dont send actual audio bytes in event message - cJSON* jsonFile = NULL; - cJSON* jsonAudio = cJSON_DetachItemFromObject(jsonData, "audioContent"); - int validAudio = (jsonAudio && NULL != jsonAudio->valuestring); - - const char* szAudioContentType = cJSON_GetObjectCstr(jsonData, "audioContentType"); - char fileType[6]; - int sampleRate = 16000; - if (0 == strcmp(szAudioContentType, "raw")) { - cJSON* jsonSR = cJSON_GetObjectItem(jsonData, "sampleRate"); - sampleRate = jsonSR && jsonSR->valueint ? jsonSR->valueint : 0; - - switch(sampleRate) { - case 8000: - strcpy(fileType, ".r8"); - break; - case 16000: - strcpy(fileType, ".r16"); - break; - case 24000: - strcpy(fileType, ".r24"); - break; - case 32000: - strcpy(fileType, ".r32"); - break; - case 48000: - strcpy(fileType, ".r48"); - break; - case 64000: - strcpy(fileType, ".r64"); - break; - default: - strcpy(fileType, ".r16"); - break; - } - } - else if (0 == strcmp(szAudioContentType, "wave") || 0 == strcmp(szAudioContentType, "wav")) { - strcpy(fileType, ".wav"); - } - else { - validAudio = 0; - switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) processIncomingMessage - unsupported audioContentType: %s\n", tech_pvt->id, szAudioContentType); - } - - if (validAudio) { - char szFilePath[256]; - - std::string rawAudio = drachtio::base64_decode(jsonAudio->valuestring); - switch_snprintf(szFilePath, 256, "%s%s%s_%d.tmp%s", SWITCH_GLOBAL_dirs.temp_dir, - SWITCH_PATH_SEPARATOR, tech_pvt->sessionId, playCount++, fileType); - std::ofstream f(szFilePath, std::ofstream::binary); - f << rawAudio; - f.close(); - - // add the file to the list of files played for this session, we'll delete when session closes - struct playout* playout = (struct playout *) malloc(sizeof(struct playout)); - playout->file = (char *) malloc(strlen(szFilePath) + 1); - strcpy(playout->file, szFilePath); - playout->next = tech_pvt->playout; - tech_pvt->playout = playout; - - jsonFile = cJSON_CreateString(szFilePath); - cJSON_AddItemToObject(jsonData, "file", jsonFile); - } - - char* jsonString = cJSON_PrintUnformatted(jsonData); - tech_pvt->responseHandler(session, EVENT_PLAY_AUDIO, jsonString); - free(jsonString); - if (jsonAudio) cJSON_Delete(jsonAudio); - } - else { - switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "(%u) processIncomingMessage - missing data payload in playAudio request\n", tech_pvt->id); - } - } - else if (0 == type.compare("killAudio")) { - tech_pvt->responseHandler(session, EVENT_KILL_AUDIO, NULL); + if (0 == type.compare("killAudio")) { + tech_pvt->responseHandler(session, tech_pvt->bugname, EVENT_KILL_AUDIO, NULL); // kill any current playback on the channel switch_channel_t *channel = switch_core_session_get_channel(session); switch_channel_set_flag_value(channel, CF_BREAK, 2); // this will dump buffered incoming audio - tech_pvt->clear_bidirectional_audio_buffer = true; - } - else if (0 == type.compare("transcription")) { - char* jsonString = cJSON_PrintUnformatted(jsonData); - tech_pvt->responseHandler(session, EVENT_TRANSCRIPTION, jsonString); - free(jsonString); + ss->clearPlayout = true; } else if (0 == type.compare("transfer")) { char* jsonString = cJSON_PrintUnformatted(jsonData); - tech_pvt->responseHandler(session, EVENT_TRANSFER, jsonString); + tech_pvt->responseHandler(session, tech_pvt->bugname, EVENT_TRANSFER, jsonString); free(jsonString); } else if (0 == type.compare("disconnect")) { char* jsonString = cJSON_PrintUnformatted(jsonData); - tech_pvt->responseHandler(session, EVENT_DISCONNECT, jsonString); + tech_pvt->responseHandler(session, tech_pvt->bugname, EVENT_DISCONNECT, jsonString); free(jsonString); } else if (0 == type.compare("error")) { char* jsonString = cJSON_PrintUnformatted(jsonData); - tech_pvt->responseHandler(session, EVENT_ERROR, jsonString); + tech_pvt->responseHandler(session, tech_pvt->bugname, EVENT_ERROR, jsonString); free(jsonString); } else if (0 == type.compare("json")) { char* jsonString = cJSON_PrintUnformatted(json); - tech_pvt->responseHandler(session, EVENT_JSON, jsonString); + tech_pvt->responseHandler(session, tech_pvt->bugname, EVENT_JSON, jsonString); free(jsonString); } else { @@ -304,7 +273,8 @@ namespace { } } - static void eventCallback(const char* sessionId, const char* bugname, drachtio::AudioPipe::NotifyEvent_t event, const char* message, const char* binary, size_t len) { + static void eventCallback(const char* sessionId, const char* bugname, drachtio::AudioPipe::NotifyEvent_t event, const char* message, const char* binary, size_t len, void* userData) { + StreamState* ss = static_cast(userData); switch_core_session_t* session = switch_core_session_locate(sessionId); if (session) { switch_channel_t *channel = switch_core_session_get_channel(session); @@ -315,39 +285,28 @@ namespace { switch (event) { case drachtio::AudioPipe::CONNECT_SUCCESS: switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "connection successful\n"); - tech_pvt->responseHandler(session, EVENT_CONNECT_SUCCESS, NULL); - if (strlen(tech_pvt->initialMetadata) > 0) { - switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "sending initial metadata %s\n", tech_pvt->initialMetadata); - drachtio::AudioPipe *pAudioPipe = static_cast(tech_pvt->pAudioPipe); - pAudioPipe->bufferForSending(tech_pvt->initialMetadata); - } + tech_pvt->responseHandler(session, tech_pvt->bugname, EVENT_CONNECT_SUCCESS, NULL); break; case drachtio::AudioPipe::CONNECT_FAIL: { - // first thing: we can no longer access the AudioPipe std::stringstream json; json << "{\"reason\":\"" << message << "\"}"; - tech_pvt->pAudioPipe = nullptr; - tech_pvt->responseHandler(session, EVENT_CONNECT_FAIL, (char *) json.str().c_str()); + tech_pvt->responseHandler(session, tech_pvt->bugname, EVENT_CONNECT_FAIL, (char *) json.str().c_str()); switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_NOTICE, "connection failed: %s\n", message); } break; case drachtio::AudioPipe::CONNECTION_DROPPED: - // first thing: we can no longer access the AudioPipe - tech_pvt->pAudioPipe = nullptr; - tech_pvt->responseHandler(session, EVENT_DISCONNECT, NULL); + tech_pvt->responseHandler(session, tech_pvt->bugname, EVENT_DISCONNECT, NULL); switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_NOTICE, "connection dropped from far end\n"); break; case drachtio::AudioPipe::CONNECTION_CLOSED_GRACEFULLY: - // first thing: we can no longer access the AudioPipe - tech_pvt->pAudioPipe = nullptr; switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "connection closed gracefully\n"); break; case drachtio::AudioPipe::MESSAGE: - processIncomingMessage(tech_pvt, session, message); + processIncomingMessage(tech_pvt, session, message, ss); break; case drachtio::AudioPipe::BINARY: - processIncomingBinary(tech_pvt, session, binary, len); + processIncomingBinary(ss, binary, len); break; } } @@ -363,57 +322,56 @@ namespace { const char* username = nullptr; const char* password = nullptr; int err; - int bidirectional_audio_stream_enable = bidirectional_audio_enable + bidirectional_audio_stream; + // must match the condition under which start sets SMBF_WRITE_REPLACE, otherwise nothing drains the playout buffer + int bidirectional_audio_stream_enable = + bidirectional_audio_enable && bidirectional_audio_stream && bidirectional_audio_sample_rate; switch_codec_implementation_t read_impl; switch_channel_t *channel = switch_core_session_get_channel(session); - switch_core_session_get_read_impl(session, &read_impl); + // zeroed first so destroy_tech_pvt is safe on every early return below + memset(tech_pvt, 0, sizeof(private_t)); + + if (switch_core_session_get_read_impl(session, &read_impl) != SWITCH_STATUS_SUCCESS) { + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "unable to get the read codec implementation\n"); + return SWITCH_STATUS_FALSE; + } if (username = switch_channel_get_variable(channel, "MOD_AUDIO_BASIC_AUTH_USERNAME")) { password = switch_channel_get_variable(channel, "MOD_AUDIO_BASIC_AUTH_PASSWORD"); } - memset(tech_pvt, 0, sizeof(private_t)); - - strncpy(tech_pvt->sessionId, switch_core_session_get_uuid(session), MAX_SESSION_ID); - strncpy(tech_pvt->host, host, MAX_WS_URL_LEN); - tech_pvt->port = port; - strncpy(tech_pvt->path, path, MAX_PATH_LEN); + tech_pvt->sessionId = switch_core_session_strdup(session, switch_core_session_get_uuid(session)); tech_pvt->sampling = desiredSampling; tech_pvt->responseHandler = responseHandler; - tech_pvt->playout = NULL; tech_pvt->channels = channels; tech_pvt->id = ++idxCallCount; tech_pvt->buffer_overrun_notified = 0; tech_pvt->audio_paused = 0; tech_pvt->graceful_shutdown = 0; - tech_pvt->streamingPlayoutBuffer = (void *) new CircularBuffer_t(8192); + StreamHandle::Ptr stream = std::make_shared(); tech_pvt->bidirectional_audio_enable = bidirectional_audio_enable; tech_pvt->bidirectional_audio_stream = bidirectional_audio_stream; tech_pvt->bidirectional_audio_sample_rate = bidirectional_audio_sample_rate; - tech_pvt->clear_bidirectional_audio_buffer = false; - tech_pvt->has_set_aside_byte = 0; - tech_pvt->downscale_factor = 1; if (bidirectional_audio_sample_rate > sampling) { - tech_pvt->downscale_factor = bidirectional_audio_sample_rate / sampling; - switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "downscale_factor is %d\n", tech_pvt->downscale_factor); + stream->downscaleFactor = bidirectional_audio_sample_rate / sampling; + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "downscale_factor is %d\n", stream->downscaleFactor); } - tech_pvt->streamingPreBufSize = 320 * tech_pvt->downscale_factor * 4; // min 80ms prebuffer - tech_pvt->streamingPreBuffer = (void *) new CircularBuffer_t(8192); + stream->preBufSize = 320 * stream->downscaleFactor * 4; // min 80ms prebuffer - strncpy(tech_pvt->bugname, bugname, MAX_BUG_LEN); - if (metadata) strncpy(tech_pvt->initialMetadata, metadata, MAX_METADATA_LEN); + tech_pvt->bugname = switch_core_session_strdup(session, bugname); size_t buflen = LWS_PRE + (FRAME_SIZE_8000 * desiredSampling / 8000 * channels * 1000 / RTP_PACKETIZATION_PERIOD * nAudioBufferSecs); - drachtio::AudioPipe* ap = new drachtio::AudioPipe(tech_pvt->sessionId, host, port, path, sslFlags, - buflen, read_impl.decoded_bytes_per_packet, username, password, bugname, bidirectional_audio_stream_enable, eventCallback); - if (!ap) { - switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error allocating AudioPipe\n"); + PipeHandle::Ptr ap = std::make_shared(tech_pvt->sessionId, host, port, path, sslFlags, + buflen, read_impl.decoded_bytes_per_packet, username, password, bugname, bidirectional_audio_stream_enable, eventCallback, stream); + + if (metadata && *metadata && !ap->enqueueText(metadata)) { + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "metadata exceeds max length of %d bytes\n", MAX_TEXT_LEN); return SWITCH_STATUS_FALSE; } - tech_pvt->pAudioPipe = static_cast(ap); + PipeHandle::set(tech_pvt, ap); + StreamHandle::set(tech_pvt, stream); switch_mutex_init(&tech_pvt->mutex, SWITCH_MUTEX_NESTED, switch_core_session_get_pool(session)); @@ -431,7 +389,7 @@ namespace { if (bidirectional_audio_sample_rate && sampling != bidirectional_audio_sample_rate) { switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) bidirectional audio resampling from %u to %u, channels %d\n", tech_pvt->id, bidirectional_audio_sample_rate, sampling, channels); - tech_pvt->bidirectional_audio_resampler = speex_resampler_init(1, bidirectional_audio_sample_rate, sampling, SWITCH_RESAMPLE_QUALITY, &err); + stream->resampler = speex_resampler_init(1, bidirectional_audio_sample_rate, sampling, SWITCH_RESAMPLE_QUALITY, &err); if (0 != err) { switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error initializing bidirectional audio resampler: %s.\n", speex_resampler_strerror(err)); return SWITCH_STATUS_FALSE; @@ -445,28 +403,15 @@ namespace { void destroy_tech_pvt(private_t* tech_pvt) { switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "%s (%u) destroy_tech_pvt\n", tech_pvt->sessionId, tech_pvt->id); + // drop the session's references; the pipe and the stream state are freed once the lws side is done with them too + PipeHandle::release(tech_pvt); + StreamHandle::release(tech_pvt); if (tech_pvt->resampler) { speex_resampler_destroy(tech_pvt->resampler); tech_pvt->resampler = nullptr; } - if (tech_pvt->bidirectional_audio_resampler) { - speex_resampler_destroy(tech_pvt->bidirectional_audio_resampler); - tech_pvt->bidirectional_audio_resampler = nullptr; - } - if (tech_pvt->mutex) { - switch_mutex_destroy(tech_pvt->mutex); - tech_pvt->mutex = nullptr; - } - if (tech_pvt->streamingPlayoutBuffer) { - CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->streamingPlayoutBuffer; - delete cBuffer; - tech_pvt->streamingPlayoutBuffer = nullptr; - } - if (tech_pvt->streamingPreBuffer) { - CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->streamingPreBuffer; - delete cBuffer; - tech_pvt->streamingPreBuffer = nullptr; - } + // tech_pvt->mutex is not destroyed: it lives in the session pool, and a second fork_session_cleanup + // may still be waiting on it to find out that the first one is done } void lws_logger(int level, const char *line) { @@ -486,7 +431,6 @@ namespace { extern "C" { int parse_ws_uri(switch_channel_t *channel, const char* szServerUri, char* host, char *path, unsigned int* pPort, int* pSslFlags) { int i = 0, offset; - char server[MAX_WS_URL_LEN + MAX_PATH_LEN]; char *saveptr; int flags = LCCSCF_USE_SSL; @@ -504,7 +448,7 @@ extern "C" { } // get the scheme - strncpy(server, szServerUri, MAX_WS_URL_LEN + MAX_PATH_LEN); + const char *server = szServerUri; if (0 == strncmp(server, "https://", 8) || 0 == strncmp(server, "HTTPS://", 8)) { *pSslFlags = flags; offset = 8; @@ -530,30 +474,42 @@ extern "C" { return 0; } - std::string strHost(server + offset); - //- `([^/:]+)` captures the hostname/IP address, match any character except in the set - //- `:?([0-9]*)?` optionally captures a colon and the port number, if it's present. - //- `(/.*)` captures everything else (the path). - std::regex re("([^/:]+):?([0-9]*)?(/.*)?$"); - std::smatch matches; - if(std::regex_search(strHost, matches, re)) { - /* - for (int i = 0; i < matches.length(); i++) { - switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "parse_ws_uri - %d: %s\n", i, matches[i].str().c_str()); - } - */ - strncpy(host, matches[1].str().c_str(), MAX_WS_URL_LEN); - if (matches[2].str().length() > 0) { - *pPort = atoi(matches[2].str().c_str()); - } - if (matches[3].str().length() > 0) { - strncpy(path, matches[3].str().c_str(), MAX_PATH_LEN); - } - else { - strcpy(path, "/"); + // called from C: std::regex and std::string can throw (out of memory, pattern too complex for the input) + try { + std::string strHost(server + offset); + //- `([^/:]+)` captures the hostname/IP address, match any character except in the set + //- `:?([0-9]*)?` optionally captures a colon and the port number, if it's present. + //- `(/.*)` captures everything else (the path). + std::regex re("([^/:]+):?([0-9]*)?(/.*)?$"); + std::smatch matches; + if(std::regex_search(strHost, matches, re)) { + /* + for (int i = 0; i < matches.length(); i++) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "parse_ws_uri - %d: %s\n", i, matches[i].str().c_str()); + } + */ + const std::string strMatchedHost = matches[1].str(); + const std::string strMatchedPath = matches[3].str(); + if (strMatchedHost.length() >= MAX_WS_URL_LEN || strMatchedPath.length() >= MAX_PATH_LEN) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "parse_ws_uri - host or path too long\n"); + return 0; + } + switch_copy_string(host, strMatchedHost.c_str(), MAX_WS_URL_LEN); + if (matches[2].str().length() > 0) { + *pPort = atoi(matches[2].str().c_str()); + } + if (matches[3].str().length() > 0) { + switch_copy_string(path, strMatchedPath.c_str(), MAX_PATH_LEN); + } + else { + switch_copy_string(path, "/", MAX_PATH_LEN); + } + } else { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "parse_ws_uri - invalid format %s\n", strHost.c_str()); + return 0; } - } else { - switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "parse_ws_uri - invalid format %s\n", strHost.c_str()); + } catch (...) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "parse_ws_uri - error parsing uri %s\n", szServerUri); return 0; } switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "parse_ws_uri - host %s, path %s\n", host, path); @@ -610,8 +566,17 @@ extern "C" { return SWITCH_STATUS_FALSE; } - if (SWITCH_STATUS_SUCCESS != fork_data_init(tech_pvt, session, host, port, path, sslFlags, samples_per_second, sampling, channels, - bugname, metadata, bidirectional_audio_enable, bidirectional_audio_stream, bidirectional_audio_sample_rate, responseHandler)) { + // this is called from C: an exception (out of memory) must not leave this function + switch_status_t status = SWITCH_STATUS_FALSE; + try { + status = fork_data_init(tech_pvt, session, host, port, path, sslFlags, samples_per_second, sampling, channels, + bugname, metadata, bidirectional_audio_enable, bidirectional_audio_stream, bidirectional_audio_sample_rate, responseHandler); + } catch (const std::exception& e) { + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "fork_session_init failed: %s\n", e.what()); + } catch (...) { + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "fork_session_init failed\n"); + } + if (SWITCH_STATUS_SUCCESS != status) { destroy_tech_pvt(tech_pvt); return SWITCH_STATUS_FALSE; } @@ -620,9 +585,17 @@ extern "C" { return SWITCH_STATUS_SUCCESS; } + // for a session that was initialised but never got as far as having a media bug (or a connection) to clean up + void fork_session_destroy(void **ppUserData) { + private_t *tech_pvt = static_cast(*ppUserData); + if (tech_pvt) destroy_tech_pvt(tech_pvt); + *ppUserData = nullptr; + } + switch_status_t fork_session_connect(void **ppUserData) { private_t *tech_pvt = static_cast(*ppUserData); - drachtio::AudioPipe *pAudioPipe = static_cast(tech_pvt->pAudioPipe); + drachtio::AudioPipe *pAudioPipe = PipeHandle::get(tech_pvt); + if (!pAudioPipe) return SWITCH_STATUS_FALSE; pAudioPipe->connect(); return SWITCH_STATUS_SUCCESS; } @@ -635,40 +608,36 @@ extern "C" { return SWITCH_STATUS_FALSE; } private_t* tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug); + if (!tech_pvt) return SWITCH_STATUS_FALSE; uint32_t id = tech_pvt->id; switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) fork_session_cleanup\n", id); - if (!tech_pvt) return SWITCH_STATUS_FALSE; - drachtio::AudioPipe *pAudioPipe = static_cast(tech_pvt->pAudioPipe); - switch_mutex_lock(tech_pvt->mutex); - // get the bug again, now that we are under lock - { - switch_media_bug_t *bug = (switch_media_bug_t*) switch_channel_get_private(channel, bugname); - if (bug) { - switch_channel_set_private(channel, bugname, NULL); - if (!channelIsClosing) { - switch_core_media_bug_remove(session, &bug); - } - } + // get the bug again, now that we are under lock. stop and the media bug close can both get here + // (a close can even be triggered by the bug removal below); only the first one still finds the bug + // and tears down, any other returns here. + bug = (switch_media_bug_t*) switch_channel_get_private(channel, bugname); + if (!bug) { + switch_mutex_unlock(tech_pvt->mutex); + return SWITCH_STATUS_FALSE; } - - // delete any temp files - struct playout* playout = tech_pvt->playout; - while (playout) { - std::remove(playout->file); - free(playout->file); - struct playout *tmp = playout; - playout = playout->next; - free(tmp); + switch_channel_set_private(channel, bugname, NULL); + if (!channelIsClosing) { + switch_core_media_bug_remove(session, &bug); } + drachtio::AudioPipe *pAudioPipe = PipeHandle::get(tech_pvt); - if (pAudioPipe && text) pAudioPipe->bufferForSending(text); + // A final text is only worth sending on a live connection. Checking the state and then queueing is + // not atomic: the state may change in between, e.g. the connection drops. That is harmless, enqueueText + // re-checks the state under the queue lock and either queues the text (it is freed with the pipe + // if never sent) or rejects it, and we ignore the result because the pipe is closed right after. + if (pAudioPipe && text && pAudioPipe->getLwsState() == drachtio::AudioPipe::LWS_CLIENT_CONNECTED) pAudioPipe->enqueueText(text); if (pAudioPipe) pAudioPipe->close(); destroy_tech_pvt(tech_pvt); + switch_mutex_unlock(tech_pvt->mutex); switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "(%u) fork_session_cleanup: connection closed\n", id); return SWITCH_STATUS_SUCCESS; } @@ -683,8 +652,16 @@ extern "C" { private_t* tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug); if (!tech_pvt) return SWITCH_STATUS_FALSE; - drachtio::AudioPipe *pAudioPipe = static_cast(tech_pvt->pAudioPipe); - if (pAudioPipe && text) pAudioPipe->bufferForSending(text); + // the pipe handle is released under this mutex by fork_session_cleanup (null afterwards) + switch_mutex_lock(tech_pvt->mutex); + drachtio::AudioPipe *pAudioPipe = PipeHandle::get(tech_pvt); + bool failed = pAudioPipe && text && !pAudioPipe->enqueueText(text); + switch_mutex_unlock(tech_pvt->mutex); + + if (failed) { + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "fork_session_send_text failed: text exceeds max length of %d bytes, send queue is full, or connection is closed\n", MAX_TEXT_LEN); + return SWITCH_STATUS_FALSE; + } return SWITCH_STATUS_SUCCESS; } @@ -718,8 +695,11 @@ extern "C" { tech_pvt->graceful_shutdown = 1; - drachtio::AudioPipe *pAudioPipe = static_cast(tech_pvt->pAudioPipe); + // the pipe handle is released under this mutex by fork_session_cleanup (null afterwards) + switch_mutex_lock(tech_pvt->mutex); + drachtio::AudioPipe *pAudioPipe = PipeHandle::get(tech_pvt); if (pAudioPipe) pAudioPipe->do_graceful_shutdown(); + switch_mutex_unlock(tech_pvt->mutex); return SWITCH_STATUS_SUCCESS; } @@ -733,12 +713,8 @@ extern "C" { if (!tech_pvt || tech_pvt->audio_paused || tech_pvt->graceful_shutdown) return SWITCH_TRUE; if (switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) { - if (!tech_pvt->pAudioPipe) { - switch_mutex_unlock(tech_pvt->mutex); - return SWITCH_TRUE; - } - drachtio::AudioPipe *pAudioPipe = static_cast(tech_pvt->pAudioPipe); - if (pAudioPipe->getLwsState() != drachtio::AudioPipe::LWS_CLIENT_CONNECTED) { + drachtio::AudioPipe *pAudioPipe = PipeHandle::get(tech_pvt); + if (!pAudioPipe || pAudioPipe->getLwsState() != drachtio::AudioPipe::LWS_CLIENT_CONNECTED) { switch_mutex_unlock(tech_pvt->mutex); return SWITCH_TRUE; } @@ -755,7 +731,7 @@ extern "C" { if (available < pAudioPipe->binaryMinSpace()) { if (!tech_pvt->buffer_overrun_notified) { tech_pvt->buffer_overrun_notified = 1; - tech_pvt->responseHandler(session, EVENT_BUFFER_OVERRUN, NULL); + tech_pvt->responseHandler(session, tech_pvt->bugname, EVENT_BUFFER_OVERRUN, NULL); } switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "(%u) dropping packets!\n", tech_pvt->id); @@ -782,7 +758,7 @@ extern "C" { frame.buflen = SWITCH_RECOMMENDED_BUFFER_SIZE; while (switch_core_media_bug_read(bug, &frame, SWITCH_TRUE) == SWITCH_STATUS_SUCCESS) { if (frame.datalen) { - spx_uint32_t out_len = available >> 1; // space for samples which are 2 bytes + spx_uint32_t out_len = available / (sizeof(spx_int16_t) * tech_pvt->channels); // space for frames of 2 bytes per channel spx_uint32_t in_len = frame.samples; speex_resampler_process_interleaved_int(tech_pvt->resampler, @@ -793,7 +769,7 @@ extern "C" { if (out_len > 0) { // bytes written = num samples * 2 * num channels - size_t bytes_written = out_len << tech_pvt->channels; + size_t bytes_written = out_len * sizeof(spx_int16_t) * tech_pvt->channels; pAudioPipe->binaryWritePtrAdd(bytes_written); available = pAudioPipe->binarySpaceAvailable(); dirty = true; @@ -803,7 +779,7 @@ extern "C" { tech_pvt->buffer_overrun_notified = 1; switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "(%u) dropping packets!\n", tech_pvt->id); - tech_pvt->responseHandler(session, EVENT_BUFFER_OVERRUN, NULL); + tech_pvt->responseHandler(session, tech_pvt->bugname, EVENT_BUFFER_OVERRUN, NULL); } break; } @@ -817,40 +793,54 @@ extern "C" { return SWITCH_TRUE; } + // Write-replace callback of the media bug, called on the media thread for every outgoing frame (20 ms). + // Mixes the audio streamed in from the websocket (queued in StreamState::playout by processIncomingBinary + // on the lws thread) into the frame that is sent to the caller: it takes up to one frame of samples from + // the playout buffer, adds them to the frame and writes the frame back. If a killAudio cleared the buffer + // (clearPlayout), the buffered audio is dropped instead and this frame is left unchanged. + // Never blocks and never fails: when the session mutex or the playout buffer is busy, or the stream state + // is already released, the frame is passed on untouched and the buffered audio is played with the next frame. switch_bool_t dub_speech_frame(switch_media_bug_t *bug, private_t* tech_pvt) { - CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->streamingPlayoutBuffer; if (switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) { + // the stream state is released under tech_pvt->mutex by cleanup, so it is read under it here + StreamState* ss = StreamHandle::get(tech_pvt); + // like the session lock, the playout buffer is only tried: skip this frame rather than wait for the lws thread + std::unique_lock playoutLock; + if (ss) playoutLock = std::unique_lock(ss->playoutMutex, std::try_to_lock); + if (ss && playoutLock.owns_lock()) { + CircularBuffer_t *cBuffer = &ss->playout; + + // if flag was set to clear the buffer, do so and clear the flag + if (ss->clearPlayout.exchange(false)) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - clearing buffer\n", tech_pvt->id); + cBuffer->clear(); + } + else { + switch_frame_t* rframe = switch_core_media_bug_get_write_replace_frame(bug); + int16_t *fp = reinterpret_cast(rframe->data); - // if flag was set to clear the buffer, do so and clear the flag - if (tech_pvt->clear_bidirectional_audio_buffer) { - switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - clearing buffer\n", tech_pvt->id); - cBuffer->clear(); - tech_pvt->clear_bidirectional_audio_buffer = false; - } - else { - switch_frame_t* rframe = switch_core_media_bug_get_write_replace_frame(bug); - int16_t *fp = reinterpret_cast(rframe->data); + rframe->channels = 1; + rframe->datalen = rframe->samples * sizeof(int16_t); - rframe->channels = 1; - rframe->datalen = rframe->samples * sizeof(int16_t); + int16_t data[SWITCH_RECOMMENDED_BUFFER_SIZE]; + memset(data, 0, sizeof(data)); - int16_t data[SWITCH_RECOMMENDED_BUFFER_SIZE]; - memset(data, 0, sizeof(data)); + int samplesToCopy = std::min(static_cast(cBuffer->size()), static_cast(rframe->samples)); - int samplesToCopy = std::min(static_cast(cBuffer->size()), static_cast(rframe->samples)); + //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - samples to copy %u\n", tech_pvt->id, samplesToCopy); - //switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "(%u) dub_speech_frame - samples to copy %u\n", tech_pvt->id, samplesToCopy); + std::copy_n(cBuffer->begin(), samplesToCopy, data); + cBuffer->erase(cBuffer->begin(), cBuffer->begin() + samplesToCopy); - std::copy_n(cBuffer->begin(), samplesToCopy, data); - cBuffer->erase(cBuffer->begin(), cBuffer->begin() + samplesToCopy); + if (samplesToCopy > 0) { + vector_add(fp, data, rframe->samples); + } + vector_normalize(fp, rframe->samples); - if (samplesToCopy > 0) { - vector_add(fp, data, rframe->samples); + switch_core_media_bug_set_write_replace_frame(bug, rframe); } - vector_normalize(fp, rframe->samples); - - switch_core_media_bug_set_write_replace_frame(bug, rframe); } + playoutLock = std::unique_lock(); switch_mutex_unlock(tech_pvt->mutex); } return SWITCH_TRUE; @@ -865,11 +855,11 @@ extern "C" { } private_t* tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug); - CircularBuffer_t *cBuffer = (CircularBuffer_t *) tech_pvt->streamingPlayoutBuffer; - if (switch_mutex_trylock(tech_pvt->mutex) == SWITCH_STATUS_SUCCESS) { - if (cBuffer != nullptr) { - cBuffer->clear(); + StreamState* ss = StreamHandle::get(tech_pvt); + if (ss) { + std::lock_guard lk(ss->playoutMutex); + ss->playout.clear(); } switch_mutex_unlock(tech_pvt->mutex); } diff --git a/mod_audio_fork/lws_glue.h b/mod_audio_fork/lws_glue.h index e4dfb26..ad60271 100644 --- a/mod_audio_fork/lws_glue.h +++ b/mod_audio_fork/lws_glue.h @@ -20,4 +20,5 @@ switch_bool_t fork_frame(switch_core_session_t *session, switch_media_bug_t *bug switch_bool_t dub_speech_frame(switch_media_bug_t *bug, private_t * tech_pvt); switch_status_t fork_service_threads(); switch_status_t fork_session_connect(void **ppUserData); +void fork_session_destroy(void **ppUserData); #endif diff --git a/mod_audio_fork/mod_audio_fork.c b/mod_audio_fork/mod_audio_fork.c index e0a5017..0fcba15 100644 --- a/mod_audio_fork/mod_audio_fork.c +++ b/mod_audio_fork/mod_audio_fork.c @@ -14,13 +14,14 @@ SWITCH_MODULE_LOAD_FUNCTION(mod_audio_fork_load); SWITCH_MODULE_DEFINITION(mod_audio_fork, mod_audio_fork_load, mod_audio_fork_shutdown, NULL /*mod_audio_fork_runtime*/); -static void responseHandler(switch_core_session_t* session, const char * eventName, char * json) { +static void responseHandler(switch_core_session_t* session, const char * bugname, const char * eventName, char * json) { switch_event_t *event; switch_channel_t *channel = switch_core_session_get_channel(session); if (json) switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "responseHandler: sending event payload: %s.\n", json); switch_event_create_subclass(&event, SWITCH_EVENT_CUSTOM, eventName); switch_channel_event_set_data(channel, event); + switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Audio-Fork-Bugname", bugname); if (json) switch_event_add_body(event, "%s", json); switch_event_fire(&event); } @@ -87,13 +88,18 @@ static switch_status_t start_capture(switch_core_session_t *session, return SWITCH_STATUS_FALSE; } - read_codec = switch_core_session_get_read_codec(session); - if (switch_channel_pre_answer(channel) != SWITCH_STATUS_SUCCESS) { switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "mod_audio_fork: channel must have reached pre-answer status before calling start!\n"); return SWITCH_STATUS_FALSE; } + /* only available once media is set up, so ask after pre-answer */ + read_codec = switch_core_session_get_read_codec(session); + if (!read_codec || !read_codec->implementation) { + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "mod_audio_fork: channel has no read codec yet, cannot start!\n"); + return SWITCH_STATUS_FALSE; + } + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "calling fork_session_init.\n"); if (SWITCH_STATUS_FALSE == fork_session_init(session, responseHandler, read_codec->implementation->actual_samples_per_second, host, port, path, sampling, sslFlags, channels, bugname, metadata, bidirectional_audio_enable, bidirectional_audio_stream, @@ -104,6 +110,7 @@ static switch_status_t start_capture(switch_core_session_t *session, switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "adding bug %s.\n", bugname); if ((status = switch_core_media_bug_add(session, bugname, NULL, capture_callback, pUserData, 0, flags, &bug)) != SWITCH_STATUS_SUCCESS) { + fork_session_destroy(&pUserData); return status; } switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "setting bug private data %s.\n", bugname); @@ -111,6 +118,7 @@ static switch_status_t start_capture(switch_core_session_t *session, if (fork_session_connect(&pUserData) != SWITCH_STATUS_SUCCESS) { switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error mod_audio_fork session cannot connect.\n"); + fork_session_cleanup(session, bugname, NULL, 0); return SWITCH_STATUS_FALSE; } @@ -170,16 +178,16 @@ static switch_status_t send_text(switch_core_session_t *session, char* bugname, switch_media_bug_t *bug = switch_channel_get_private(channel, bugname); if (bug) { - switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "mod_audio_fork (%s): sending text: %s.\n", bugname, text); + // switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "mod_audio_fork (%s): sending text: %s.\n", bugname, text); status = fork_session_send_text(session, bugname, text); } else { - switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "mod_audio_fork (%s): no bug, failed sending text: %s.\n", bugname, text); + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "mod_audio_fork (%s): no bug, failed sending text.\n", bugname); } return status; } -#define FORK_API_SYNTAX " [start | stop | send_text | pause | resume | graceful-shutdown | stop_play ] [wss-url | path] [mono | mixed | stereo] [8000 | 16000 | 24000 | 32000 | 64000] [bugname] [metadata] [bidirectionalAudio_enabled] [bidirectionalAudio_stream_enabled] [bidirectionalAudio_stream_samplerate]" +#define FORK_API_SYNTAX " [start | stop | send_text | pause | resume | graceful-shutdown | stop_play ] [wss-url | path] [mono | mixed | stereo] [8000 | 16000 | 24000 | 32000 | 64000] [bugname] [metadata] [bidirectionalAudio_enabled | playback] [bidirectionalAudio_stream_enabled] [bidirectionalAudio_stream_samplerate]" SWITCH_STANDARD_API(fork_function) { char *mycmd = NULL, *argv[10] = { 0 }; @@ -191,11 +199,11 @@ SWITCH_STANDARD_API(fork_function) argc = switch_separate_string(mycmd, ' ', argv, (sizeof(argv) / sizeof(argv[0]))); } assert(cmd); - switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "mod_audio_fork cmd: %s\n", cmd); + // switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "mod_audio_fork cmd: %s\n", cmd); if (zstr(cmd) || argc < 2 || - (0 == strcmp(argv[1], "start") && argc < 4)) { + (0 == strcmp(argv[1], "start") && argc < 5)) { /* start needs start */ switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error with command %s %s %s.\n", cmd, argv[0], argv[1]); stream->write_function(stream, "-USAGE: %s\n", FORK_API_SYNTAX); @@ -250,16 +258,17 @@ SWITCH_STANDARD_API(fork_function) } else if (!strcasecmp(argv[1], "start")) { switch_channel_t *channel = switch_core_session_get_channel(lsession); - char host[MAX_WS_URL_LEN], path[MAX_PATH_LEN]; + char host[MAX_WS_URL_LEN] = {0}, path[MAX_PATH_LEN] = {0}; unsigned int port; int sslFlags; int sampling = 8000; switch_media_bug_flag_t flags = SMBF_READ_STREAM; char *metadata = NULL; - int bidirectional_audio_enable = 1; + int bidirectional_audio_enable = 0; int bidirectional_audio_stream = 0; int bidirectional_audio_sample_rate = 0; + int playback_only = 0; // Expecting that bidirectional audio params is always received together with bugname and metadata even they are empty string if (argc > 9) { if (argv[5][0] != '\0') { @@ -268,7 +277,9 @@ SWITCH_STANDARD_API(fork_function) if (argv[6][0] != '\0') { metadata = argv[6]; } - bidirectional_audio_enable = !strcmp(argv[7], "true") ? 1 : 0; + // "playback": audio from the websocket is played to the caller, caller audio is not forked + playback_only = !strcmp(argv[7], "playback"); + bidirectional_audio_enable = (playback_only || !strcmp(argv[7], "true")) ? 1 : 0; bidirectional_audio_stream = !strcmp(argv[8], "true") ? 1 : 0; bidirectional_audio_sample_rate = atoi(argv[9]); @@ -276,6 +287,7 @@ SWITCH_STANDARD_API(fork_function) bidirectional_audio_stream && bidirectional_audio_sample_rate) { flags |= SMBF_WRITE_REPLACE ; + if (playback_only) flags &= ~SMBF_READ_STREAM; } } else if( argc > 6 ) { bugname = argv[5]; @@ -308,11 +320,20 @@ SWITCH_STANDARD_API(fork_function) sampling = atoi(argv[4]); } if (!parse_ws_uri(channel, argv[2], &host[0], &path[0], &port, &sslFlags)) { - switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "invalid websocket uri: %s\n", argv[2]); + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(lsession), SWITCH_LOG_ERROR, "invalid websocket uri: %s\n", argv[2]); + switch_core_session_rwunlock(lsession); + goto done; + } + if (sampling <= 0 || sampling > MAX_SAMPLE_RATE || sampling % 8000 != 0) { + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(lsession), SWITCH_LOG_ERROR, "invalid sample rate: %s\n", argv[4]); + switch_core_session_rwunlock(lsession); + goto done; + } + if (bidirectional_audio_sample_rate < 0 || bidirectional_audio_sample_rate > MAX_SAMPLE_RATE) { + switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(lsession), SWITCH_LOG_ERROR, "invalid bidirectional audio sample rate: %d\n", bidirectional_audio_sample_rate); + switch_core_session_rwunlock(lsession); + goto done; } - else if (sampling % 8000 != 0) { - switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "invalid sample rate: %s\n", argv[4]); - } status = start_capture(lsession, flags, host, port, path, sampling, sslFlags, bidirectional_audio_enable, bidirectional_audio_stream, bidirectional_audio_sample_rate, bugname, metadata); } @@ -349,9 +370,7 @@ SWITCH_MODULE_LOAD_FUNCTION(mod_audio_fork_load) *module_interface = switch_loadable_module_create_module_interface(pool, modname); /* create/register custom event message types */ - if (switch_event_reserve_subclass(EVENT_TRANSCRIPTION) != SWITCH_STATUS_SUCCESS || - switch_event_reserve_subclass(EVENT_TRANSFER) != SWITCH_STATUS_SUCCESS || - switch_event_reserve_subclass(EVENT_PLAY_AUDIO) != SWITCH_STATUS_SUCCESS || + if (switch_event_reserve_subclass(EVENT_TRANSFER) != SWITCH_STATUS_SUCCESS || switch_event_reserve_subclass(EVENT_KILL_AUDIO) != SWITCH_STATUS_SUCCESS || switch_event_reserve_subclass(EVENT_ERROR) != SWITCH_STATUS_SUCCESS || switch_event_reserve_subclass(EVENT_DISCONNECT) != SWITCH_STATUS_SUCCESS) { @@ -381,9 +400,7 @@ SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_audio_fork_shutdown) { fork_cleanup(); //mod_running = 0; - switch_event_free_subclass(EVENT_TRANSCRIPTION); switch_event_free_subclass(EVENT_TRANSFER); - switch_event_free_subclass(EVENT_PLAY_AUDIO); switch_event_free_subclass(EVENT_KILL_AUDIO); switch_event_free_subclass(EVENT_DISCONNECT); switch_event_free_subclass(EVENT_ERROR); diff --git a/mod_audio_fork/mod_audio_fork.h b/mod_audio_fork/mod_audio_fork.h index ec4d120..d56fbaa 100644 --- a/mod_audio_fork/mod_audio_fork.h +++ b/mod_audio_fork/mod_audio_fork.h @@ -8,14 +8,12 @@ #include #define MY_BUG_NAME "audio_fork" -#define MAX_BUG_LEN (64) -#define MAX_SESSION_ID (256) #define MAX_WS_URL_LEN (512) #define MAX_PATH_LEN (4096) +// upper bound for the sample rates given to start (the audio buffer size grows with the rate) +#define MAX_SAMPLE_RATE (64000) -#define EVENT_TRANSCRIPTION "mod_audio_fork::transcription" #define EVENT_TRANSFER "mod_audio_fork::transfer" -#define EVENT_PLAY_AUDIO "mod_audio_fork::play_audio" #define EVENT_KILL_AUDIO "mod_audio_fork::kill_audio" #define EVENT_DISCONNECT "mod_audio_fork::disconnect" #define EVENT_ERROR "mod_audio_fork::error" @@ -24,47 +22,38 @@ #define EVENT_BUFFER_OVERRUN "mod_audio_fork::buffer_overrun" #define EVENT_JSON "mod_audio_fork::json" -#define MAX_METADATA_LEN (8192) +// Max length in bytes of the start metadata and of each send_text message. +// Both used to be limited/unbounded by accident: the start metadata was copied into a fixed +// 8K char array (silently truncated), and send_text text was unbounded and copied into a +// stack array when written, which could overflow the stack for large messages. +// Now both are held on the heap and rejected (never truncated, as truncated JSON is invalid) +// when longer than this limit. +#define MAX_TEXT_LEN (1024 * 1024) -struct playout { - char *file; - struct playout* next; -}; - -typedef void (*responseHandler_t)(switch_core_session_t* session, const char* eventName, char* json); +typedef void (*responseHandler_t)(switch_core_session_t* session, const char* bugname, const char* eventName, char* json); struct private_data { switch_mutex_t *mutex; - char sessionId[MAX_SESSION_ID]; - char bugname[MAX_BUG_LEN+1]; + const char *sessionId; + const char *bugname; SpeexResamplerState *resampler; responseHandler_t responseHandler; void *pAudioPipe; int ws_state; - char host[MAX_WS_URL_LEN]; - unsigned int port; - char path[MAX_PATH_LEN]; int sampling; - struct playout* playout; int channels; unsigned int id; int buffer_overrun_notified:1; int audio_paused:1; int graceful_shutdown:1; - char initialMetadata[8192]; // bidirectional audio - void *streamingPlayoutBuffer; - void *streamingPreBuffer; - int streamingPreBufSize; - uint8_t set_aside_byte; - int has_set_aside_byte; - int downscale_factor; - SpeexResamplerState *bidirectional_audio_resampler; + // session-side reference to the shared StreamState (see lws_glue.cpp); the lws thread reaches the same + // object through the AudioPipe, so releasing this at cleanup does not free it under the lws thread + void *pStream; int bidirectional_audio_enable; int bidirectional_audio_stream; int bidirectional_audio_sample_rate; - int clear_bidirectional_audio_buffer; }; typedef struct private_data private_t; diff --git a/mod_audio_fork/test/test_units.cpp b/mod_audio_fork/test/test_units.cpp new file mode 100644 index 0000000..f446eba --- /dev/null +++ b/mod_audio_fork/test/test_units.cpp @@ -0,0 +1,87 @@ +// Standalone unit tests: no FreeSWITCH symbols required. +#include +#include +#include +#include +#include +#include "base64.hpp" +#include "vector_math.h" + +using namespace drachtio; + +static void test_base64() { + const std::string in = "hello, audio fork!"; + std::string enc = base64_encode(in); + assert(enc == "aGVsbG8sIGF1ZGlvIGZvcmsh"); + assert(base64_decode(enc) == in); + assert(base64_decode(base64_encode(std::string())).empty()); + for (size_t n = 0; n < 8; n++) { + std::string s(n, '\xff'); + assert(base64_decode(base64_encode(s)) == s); + } +} + +static void test_vector_add() { + std::vector a(37), b(37); + for (size_t i = 0; i < a.size(); i++) { a[i] = (int16_t)(i * 10); b[i] = (int16_t)(i * 3 - 20); } + std::vector expect(a); + for (size_t i = 0; i < a.size(); i++) expect[i] = (int16_t)(a[i] + b[i]); + vector_add(a.data(), b.data(), a.size()); + assert(a == expect); +} + +static void test_vector_add_saturates() { + // 37 samples cover the SIMD blocks and the scalar tail + std::vector a(37), b(37), expect(37); + for (size_t i = 0; i < a.size(); i++) { + if (i % 3 == 0) { a[i] = 30000; b[i] = 10000; expect[i] = 32767; } + else if (i % 3 == 1) { a[i] = -30000; b[i] = -10000; expect[i] = -32768; } + else { a[i] = 12345; b[i] = -2345; expect[i] = 10000; } + } + vector_add(a.data(), b.data(), a.size()); + assert(a == expect); +} + +static void test_vector_normalize() { + std::vector a = {0, 1, -1, 32767, -32768, 100, -100}; + std::vector expect(a); + vector_normalize(a.data(), a.size()); + assert(a == expect); +} + +// vector_math.cpp does not define vector_change_sln_volume_granular in its SSE2 branch +// (and nothing else in the module calls it), so only test it where it exists. +#ifndef USE_SSE2 +static void test_volume() { + std::vector a(33, 1000); + vector_change_sln_volume_granular(a.data(), a.size(), 0); // no-op + for (int16_t v : a) assert(v == 1000); + + vector_change_sln_volume_granular(a.data(), a.size(), 1); // +1 dB, ~x1.122 + for (int16_t v : a) assert(v >= 1121 && v <= 1123); + + std::vector loud(33, 20000); + vector_change_sln_volume_granular(loud.data(), loud.size(), 20); // saturates + for (int16_t v : loud) assert(v == 32767); + + std::vector quiet(33, 1000); + vector_change_sln_volume_granular(quiet.data(), quiet.size(), -50); // silence is mapped to 0 -> left unchanged + for (int16_t v : quiet) assert(v == 1000); + + std::vector down(33, 1000); + vector_change_sln_volume_granular(down.data(), down.size(), -10); // ~x0.316 + for (int16_t v : down) assert(v >= 315 && v <= 317); +} +#endif + +int main() { + test_base64(); + test_vector_add(); + test_vector_add_saturates(); + test_vector_normalize(); +#ifndef USE_SSE2 + test_volume(); +#endif + puts("all unit tests passed"); + return 0; +} diff --git a/mod_audio_fork/vector_math.cpp b/mod_audio_fork/vector_math.cpp index 4481ebb..8eb9e3d 100644 --- a/mod_audio_fork/vector_math.cpp +++ b/mod_audio_fork/vector_math.cpp @@ -7,6 +7,11 @@ #define SMAX 32767 #define SMIN (-32768) #define normalize_to_16bit_basic(n) if (n > SMAX) n = SMAX; else if (n < SMIN) n = SMIN; +// saturating 16 bit add, matches _mm_adds_epi16 / _mm256_adds_epi16 +static inline int16_t add_saturate_16bit(int16_t x, int16_t y) { + int32_t sum = (int32_t)x + (int32_t)y; + return (int16_t)(sum > SMAX ? SMAX : (sum < SMIN ? SMIN : sum)); +} #define normalize_volume_granular(x) if (x > GRANULAR_VOLUME_MAX) x = GRANULAR_VOLUME_MAX; if (x < -GRANULAR_VOLUME_MAX) x = -GRANULAR_VOLUME_MAX; #ifdef __cplusplus @@ -21,11 +26,11 @@ void vector_add(int16_t* a, int16_t* b, size_t len) { for (; i + 15 < len; i += 16) { __m256i va = _mm256_loadu_si256((const __m256i*)(a + i)); __m256i vb = _mm256_loadu_si256((const __m256i*)(b + i)); - __m256i vc = _mm256_add_epi16(va, vb); + __m256i vc = _mm256_adds_epi16(va, vb); _mm256_storeu_si256((__m256i*)(a + i), vc); } for (; i < len; ++i) { - a[i] += b[i]; + a[i] = add_saturate_16bit(a[i], b[i]); } } void vector_normalize(int16_t* a, size_t len) { @@ -118,11 +123,11 @@ void vector_add(int16_t* a, int16_t* b, size_t len) { for (; i + 7 < len; i += 8) { __m128i va = _mm_loadu_si128((const __m128i*)(a + i)); __m128i vb = _mm_loadu_si128((const __m128i*)(b + i)); - __m128i vc = _mm_add_epi16(va, vb); + __m128i vc = _mm_adds_epi16(va, vb); _mm_storeu_si128((__m128i*)(a + i), vc); } for (; i < len; ++i) { - a[i] += b[i]; + a[i] = add_saturate_16bit(a[i], b[i]); } } void vector_normalize(int16_t* a, size_t len) { @@ -157,7 +162,7 @@ typedef union { #pragma message("Building without vector math support") void vector_add(int16_t* a, int16_t* b, size_t len) { for (size_t i = 0; i < len; i++) { - a[i] += b[i]; + a[i] = add_saturate_16bit(a[i], b[i]); } } void vector_normalize(int16_t* a, size_t len) { diff --git a/scripts/ide-setup.sh b/scripts/ide-setup.sh new file mode 100755 index 0000000..3c60588 --- /dev/null +++ b/scripts/ide-setup.sh @@ -0,0 +1,64 @@ +#!/usr/bin/env bash +# Fetch FreeSWITCH/libwebsockets headers into .deps and write .clangd and +# .vscode/c_cpp_properties.json so the IDE can resolve includes. +# Usage: scripts/ide-setup.sh +set -euo pipefail +[ $# -eq 2 ] || { echo "usage: $0 " >&2; exit 1; } +fs_version="$1" +lws_version="$2" +root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$root" +command -v cmake >/dev/null || { echo "cmake is required"; exit 1; } +deps="$root/.deps" +mkdir -p "$deps" +[ -d "$deps/freeswitch" ] || git clone --depth 1 --branch $fs_version https://github.com/signalwire/freeswitch.git "$deps/freeswitch" +[ -d "$deps/lws" ] || git clone --depth 1 --branch $lws_version https://github.com/warmcat/libwebsockets.git "$deps/lws" +# Build-time headers FreeSWITCH normally generates via configure (LP64 values, as in the Dockerfile) +inc="$deps/freeswitch/src/include" +sed -e 's/@short_value@/short/' -e 's/@int_value@/int/' -e 's/@long_value@/long/' \ + -e 's/@size_t_value@/unsigned long/' -e 's/@ssize_t_value@/long/' \ + -e 's/@voidp_size@/8/' -e 's|@prefix@|/usr/local/freeswitch|' \ + -e 's/^@[a-z0-9_]*_fmt@$//' "$inc/switch_am_config.h.in" > "$inc/switch_am_config.h" +sed -E 's/@([A-Z_]+)@/0/g' "$inc/switch_version.h.template" > "$inc/switch_version.h" +# Configure only: generates lws_config.h, no compile needed +cmake -S "$deps/lws" -B "$deps/lws/build" -DLWS_WITH_SSL=OFF -DLWS_WITHOUT_TESTAPPS=ON \ + -DLWS_WITH_MINIMAL_EXAMPLES=OFF -DCMAKE_TRY_COMPILE_TARGET_TYPE=STATIC_LIBRARY >/dev/null +# Third-party headers from Homebrew (boost, speexdsp) +extra="" +if command -v brew >/dev/null; then + for f in boost speexdsp; do + p="$(brew --prefix "$f" 2>/dev/null || true)" + if [ -d "$p/include" ]; then extra="$extra, -I$p/include"; else echo "missing: brew install $f"; fi + done +fi +# clangd may pair an Xcode toolchain with the CLT SDK and then miss libc++; point it at the SDK's copy +cxx="" +if command -v xcrun >/dev/null; then cxx=", -isystem$(xcrun --show-sdk-path)/usr/include/c++/v1"; fi +cat > .clangd </dev/null || true)" + [ -d "$p/include" ] && incs="$incs, \"$p/include\"" +done +cat > .vscode/c_cpp_properties.json <