30161608849

Committed 25 Jul 2026 02:25PM UTC coverage: 15.303% (-0.008%) from 15.311%

Build # 30161608849

Build Type

Pull #2148

github

Committed by

web-flow

Commit Message

Merge d2b82a62f into 13afd939a

Pull Request Pull Request #2148: wasm: stream large media from Blob URLs instead of copying into RAM (Tier B)

Coverage Stats

4 of 179 new or added lines in 10 files covered. (2.23%)

7 existing lines in 5 files now uncovered.

30396 of 198627 relevant lines covered (15.3%)

983.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

4.23

/src/plugins/score-plugin-media/Media/AvStreamIO.cpp

#include "AvStreamIO.hpp"

#if SCORE_HAS_LIBAV
#include <ossia/detail/libav.hpp>

#include <QString>

#include <algorithm>
#include <cstring>
#include <memory>
#include <string>
#include <vector>

#if defined(__EMSCRIPTEN__)
#include <QtCore/private/qstdweb_p.h>
#include <QtCore/private/qwasmlocalfileengine_p.h>

#include <emscripten.h>
#include <emscripten/atomic.h>
#include <emscripten/proxying.h>
#include <emscripten/threading.h>
#include <emscripten/threading_primitives.h>
#include <emscripten/val.h>

#include <cmath>
#endif

namespace Media
{
// ffmpeg pulls 4 MB per read, served from a 16 MB read-ahead window so
// sequential demuxing round-trips to the main thread once per 16 MB.
static constexpr int av_io_buffer_size = 1 << 22;
static constexpr int64_t prefetch_size = 16ll << 20;

struct AvStreamState
{
  std::string url;
  int64_t pos{};
  int64_t size{};
  std::vector<uint8_t> cache;
  int64_t cache_off{-1};
  int64_t cache_len{};
};

bool isStreamedMediaPath(const QString& path) noexcept
{
  return path.startsWith(QLatin1String("weblocalfile:"));
}

bool isStreamedMediaPath(const std::string& path) noexcept
{
  return path.rfind("weblocalfile:", 0) == 0;
}

#if defined(__EMSCRIPTEN__)

// getFile() resolves against a process-wide singleton internally, so any handler
// instance returns the File registered at drop time.
static qstdweb::File getWebFile(const std::string& url)
{
  static QWasmFileEngineHandler handler;
  return handler.getFile(QString::fromStdString(url));
}

// Blocking read on the main thread: the Blob read suspends via JSPI, which is
// legal here because the main event loop is a promising call stack.
static int read_blob_main(const std::string& url, int64_t off, int32_t want, uint8_t* out)
{
  qstdweb::File f = getWebFile(url);
  if(f.file().isUndefined() || f.file().isNull())
    return -1;

  qstdweb::ArrayBuffer ab
      = f.slice((uint64_t)off, (uint64_t)(off + want)).arrayBuffer_sync();
  const int got = (int)ab.byteLength();
  if(got > 0)
    qstdweb::Uint8Array(ab).copyTo((char*)out);
  return got;
}

// A worker cannot touch the main-thread Blob val, and cannot suspend via JSPI.
// So it proxies an *async* read to the main thread and blocks on a futex: the
// main job kicks off Blob.arrayBuffer() (no suspend) and its .then copies the
// bytes into shared memory and wakes the worker. `state` must stay first so its
// address is the futex word.
struct AsyncRead
{
  int32_t state{}; // 0 = pending, 1 = complete
  int32_t result{}; // bytes read, or -1 on error
  const std::string* url{};
  int64_t off{};
  int32_t want{};
  uint8_t* dst{};
};

static void async_read_job(void* p)
{
  auto* a = static_cast<AsyncRead*>(p);
  qstdweb::File f = getWebFile(*a->url);
  if(f.file().isUndefined() || f.file().isNull())
  {
    a->result = -1;
    emscripten_atomic_store_u32(&a->state, 1);
    emscripten_futex_wake(&a->state, 1);
    return;
  }

  qstdweb::Blob blob = f.slice((uint64_t)a->off, (uint64_t)(a->off + a->want));
  EM_ASM(
      {
        var blob = Emval.toValue($0);
        var dst = $1;
        var statePtr = $2;
        var resultPtr = $3;
        blob.arrayBuffer()
            .then(function(ab) {
              var u8 = new Uint8Array(ab);
              HEAPU8.set(u8, dst);
              HEAP32[resultPtr >> 2] = u8.length;
              Atomics.store(HEAP32, statePtr >> 2, 1);
              Atomics.notify(HEAP32, statePtr >> 2);
            })
            .catch(function(e) {
              HEAP32[resultPtr >> 2] = -1;
              Atomics.store(HEAP32, statePtr >> 2, 1);
              Atomics.notify(HEAP32, statePtr >> 2);
            });
      },
      blob.val().as_handle(), a->dst, &a->state, &a->result);
}

static int read_blob_worker(const std::string& url, int64_t off, int32_t want, uint8_t* out)
{
  AsyncRead a;
  a.url = &url;
  a.off = off;
  a.want = want;
  a.dst = out;

  if(!emscripten_proxy_async(
         emscripten_proxy_get_system_queue(), emscripten_main_runtime_thread_id(),
         &async_read_job, &a))
    return -1;

  while(emscripten_atomic_load_u32(&a.state) == 0)
    emscripten_futex_wait(&a.state, 0, INFINITY);

  return a.result;
}

static int fetch_bytes(const std::string& url, int64_t off, int32_t want, uint8_t* dst)
{
  return emscripten_is_main_runtime_thread()
             ? read_blob_main(url, off, want, dst)
             : read_blob_worker(url, off, want, dst);
}

static int av_io_read(void* opaque, uint8_t* out, int size)
{
  auto* st = static_cast<AvStreamState*>(opaque);
  const int64_t avail = st->size - st->pos;
  if(avail <= 0)
    return AVERROR_EOF;

  const int32_t want = (int32_t)std::min<int64_t>(size, avail);

  if(st->cache_off >= 0 && st->pos >= st->cache_off
     && st->pos + want <= st->cache_off + st->cache_len)
  {
    std::memcpy(out, st->cache.data() + (st->pos - st->cache_off), want);
    st->pos += want;
    return want;
  }

  const int32_t win = (int32_t)std::min<int64_t>(prefetch_size, st->size - st->pos);
  const int got = fetch_bytes(st->url, st->pos, win, st->cache.data());
  if(got <= 0)
    return got == 0 ? AVERROR_EOF : AVERROR(EIO);
  st->cache_off = st->pos;
  st->cache_len = got;

  const int n = std::min(want, got);
  std::memcpy(out, st->cache.data(), n);
  st->pos += n;
  return n;
}

static int64_t av_io_seek(void* opaque, int64_t offset, int whence)
{
  auto* st = static_cast<AvStreamState*>(opaque);
  switch(whence & ~AVSEEK_FORCE)
  {
    case AVSEEK_SIZE:
      return st->size;
    case SEEK_SET:
      st->pos = offset;
      break;
    case SEEK_CUR:
      st->pos += offset;
      break;
    case SEEK_END:
      st->pos = st->size + offset;
      break;
    default:
      return -1;
  }
  if(st->pos < 0)
    st->pos = 0;
  return st->pos;
}

#endif

AvIoDevice::AvIoDevice(const QString& path)
{
  auto st = std::make_unique<AvStreamState>();
  st->url = path.toStdString();

#if defined(__EMSCRIPTEN__)
  bool ok = false;
  const auto fetch_size = [&] {
    qstdweb::File f = getWebFile(st->url);
    if(f.file().isUndefined() || f.file().isNull())
      return;
    st->size = (int64_t)f.size();
    ok = st->size > 0;
  };
  if(emscripten_is_main_runtime_thread())
  {
    fetch_size();
  }
  else
  {
    emscripten_proxy_sync(
        emscripten_proxy_get_system_queue(), emscripten_main_runtime_thread_id(),
        [](void* p) { (*static_cast<const decltype(fetch_size)*>(p))(); },
        const_cast<void*>(static_cast<const void*>(&fetch_size)));
  }
  if(!ok)
    return;

  st->cache.resize((size_t)std::min<int64_t>(prefetch_size, st->size));

  auto* buffer = static_cast<unsigned char*>(av_malloc(av_io_buffer_size));
  if(!buffer)
    return;

  avio = avio_alloc_context(
      buffer, av_io_buffer_size, 0, st.get(), &av_io_read, nullptr, &av_io_seek);
  if(!avio)
  {
    av_free(buffer);
    return;
  }
  state = st.release();
#endif
}

AvIoDevice::~AvIoDevice()
{
  if(avio)
  {
    av_freep(&avio->buffer);
    avio_context_free(&avio);
  }
  delete state;
  state = nullptr;
}

AvIoDevice::AvIoDevice(AvIoDevice&& other) noexcept
    : avio{other.avio}
    , state{other.state}
{
  other.avio = nullptr;
  other.state = nullptr;
}

AvIoDevice& AvIoDevice::operator=(AvIoDevice&& other) noexcept
{
  if(this != &other)
  {
    if(avio)
    {
      av_freep(&avio->buffer);
      avio_context_free(&avio);
    }
    delete state;
    avio = other.avio;
    state = other.state;
    other.avio = nullptr;
    other.state = nullptr;
  }
  return *this;
}

bool open_input_custom_io(
    AVFormatContext*& format, AvIoDevice& io, const QString& path) noexcept
{
  io = AvIoDevice{path};
  if(!io)
    return false;

  AVFormatContext* fmt = avformat_alloc_context();
  if(!fmt)
  {
    io = {};
    return false;
  }

  fmt->pb = io.avio;
  fmt->flags |= AVFMT_FLAG_CUSTOM_IO;

  if(avformat_open_input(&fmt, nullptr, nullptr, nullptr) != 0)
  {
    io = {};
    return false;
  }

  format = fmt;
  return true;
}

// libossia's libav_handle is Qt-agnostic; it delegates the browser-only Blob
// streaming back here through a hook so the submodule keeps no Qt dependency.
static AVFormatContext* libav_stream_open_hook(
    const char* path, void*& io_owner, void (*&io_free)(void*)) noexcept
{
  const QString qpath = QString::fromUtf8(path);
  if(!isStreamedMediaPath(qpath))
    return nullptr;

  auto io = std::make_unique<AvIoDevice>();
  AVFormatContext* fmt{};
  if(!open_input_custom_io(fmt, *io, qpath))
    return nullptr;

  io_owner = io.release();
  io_free = [](void* p) { delete static_cast<AvIoDevice*>(p); };
  return fmt;
}

static const bool registered_libav_hook = [] {
  ossia::libav_custom_open() = &libav_stream_open_hook;
  return true;
}();
}
#endif

1	#include "AvStreamIO.hpp"
2
3	#if SCORE_HAS_LIBAV
4	#include <ossia/detail/libav.hpp>
5
6	#include <QString>
7
8	#include <algorithm>
9	#include <cstring>
10	#include <memory>
11	#include <string>
12	#include <vector>
13
14	#if defined(__EMSCRIPTEN__)
15	#include <QtCore/private/qstdweb_p.h>
16	#include <QtCore/private/qwasmlocalfileengine_p.h>
17
18	#include <emscripten.h>
19	#include <emscripten/atomic.h>
20	#include <emscripten/proxying.h>
21	#include <emscripten/threading.h>
22	#include <emscripten/threading_primitives.h>
23	#include <emscripten/val.h>
24
25	#include <cmath>
26	#endif
27
28	namespace Media
29	{
30	// ffmpeg pulls 4 MB per read, served from a 16 MB read-ahead window so
31	// sequential demuxing round-trips to the main thread once per 16 MB.
32	static constexpr int av_io_buffer_size = 1 << 22;
33	static constexpr int64_t prefetch_size = 16ll << 20;
34
NEW 35	struct AvStreamState	×
36	{
37	std::string url;
NEW 38	int64_t pos{};	×
NEW 39	int64_t size{};	×
40	std::vector<uint8_t> cache;
NEW 41	int64_t cache_off{-1};	×
NEW 42	int64_t cache_len{};	×
43	};
44
NEW 45	bool isStreamedMediaPath(const QString& path) noexcept	×
46	{
NEW 47	return path.startsWith(QLatin1String("weblocalfile:"));	×
48	}
49
NEW 50	bool isStreamedMediaPath(const std::string& path) noexcept	×
51	{
NEW 52	return path.rfind("weblocalfile:", 0) == 0;	×
53	}
54
55	#if defined(__EMSCRIPTEN__)
56
57	// getFile() resolves against a process-wide singleton internally, so any handler
58	// instance returns the File registered at drop time.
59	static qstdweb::File getWebFile(const std::string& url)
60	{
61	static QWasmFileEngineHandler handler;
62	return handler.getFile(QString::fromStdString(url));
63	}
64
65	// Blocking read on the main thread: the Blob read suspends via JSPI, which is
66	// legal here because the main event loop is a promising call stack.
67	static int read_blob_main(const std::string& url, int64_t off, int32_t want, uint8_t* out)
68	{
69	qstdweb::File f = getWebFile(url);
70	if(f.file().isUndefined() \|\| f.file().isNull())
71	return -1;
72
73	qstdweb::ArrayBuffer ab
74	= f.slice((uint64_t)off, (uint64_t)(off + want)).arrayBuffer_sync();
75	const int got = (int)ab.byteLength();
76	if(got > 0)
77	qstdweb::Uint8Array(ab).copyTo((char*)out);
78	return got;
79	}
80
81	// A worker cannot touch the main-thread Blob val, and cannot suspend via JSPI.
82	// So it proxies an async read to the main thread and blocks on a futex: the
83	// main job kicks off Blob.arrayBuffer() (no suspend) and its .then copies the
84	// bytes into shared memory and wakes the worker. `state` must stay first so its
85	// address is the futex word.
86	struct AsyncRead
87	{
88	int32_t state{}; // 0 = pending, 1 = complete
89	int32_t result{}; // bytes read, or -1 on error
90	const std::string* url{};
91	int64_t off{};
92	int32_t want{};
93	uint8_t* dst{};
94	};
95
96	static void async_read_job(void* p)
97	{
98	auto* a = static_cast<AsyncRead*>(p);
99	qstdweb::File f = getWebFile(*a->url);
100	if(f.file().isUndefined() \|\| f.file().isNull())
101	{
102	a->result = -1;
103	emscripten_atomic_store_u32(&a->state, 1);
104	emscripten_futex_wake(&a->state, 1);
105	return;
106	}
107
108	qstdweb::Blob blob = f.slice((uint64_t)a->off, (uint64_t)(a->off + a->want));
109	EM_ASM(
110	{
111	var blob = Emval.toValue($0);
112	var dst = $1;
113	var statePtr = $2;
114	var resultPtr = $3;
115	blob.arrayBuffer()
116	.then(function(ab) {
117	var u8 = new Uint8Array(ab);
118	HEAPU8.set(u8, dst);
119	HEAP32[resultPtr >> 2] = u8.length;
120	Atomics.store(HEAP32, statePtr >> 2, 1);
121	Atomics.notify(HEAP32, statePtr >> 2);
122	})
123	.catch(function(e) {
124	HEAP32[resultPtr >> 2] = -1;
125	Atomics.store(HEAP32, statePtr >> 2, 1);
126	Atomics.notify(HEAP32, statePtr >> 2);
127	});
128	},
129	blob.val().as_handle(), a->dst, &a->state, &a->result);
130	}
131
132	static int read_blob_worker(const std::string& url, int64_t off, int32_t want, uint8_t* out)
133	{
134	AsyncRead a;
135	a.url = &url;
136	a.off = off;
137	a.want = want;
138	a.dst = out;
139
140	if(!emscripten_proxy_async(
141	emscripten_proxy_get_system_queue(), emscripten_main_runtime_thread_id(),
142	&async_read_job, &a))
143	return -1;
144
145	while(emscripten_atomic_load_u32(&a.state) == 0)
146	emscripten_futex_wait(&a.state, 0, INFINITY);
147
148	return a.result;
149	}
150
151	static int fetch_bytes(const std::string& url, int64_t off, int32_t want, uint8_t* dst)
152	{
153	return emscripten_is_main_runtime_thread()
154	? read_blob_main(url, off, want, dst)
155	: read_blob_worker(url, off, want, dst);
156	}
157
158	static int av_io_read(void* opaque, uint8_t* out, int size)
159	{
160	auto* st = static_cast<AvStreamState*>(opaque);
161	const int64_t avail = st->size - st->pos;
162	if(avail <= 0)
163	return AVERROR_EOF;
164
165	const int32_t want = (int32_t)std::min<int64_t>(size, avail);
166
167	if(st->cache_off >= 0 && st->pos >= st->cache_off
168	&& st->pos + want <= st->cache_off + st->cache_len)
169	{
170	std::memcpy(out, st->cache.data() + (st->pos - st->cache_off), want);
171	st->pos += want;
172	return want;
173	}
174
175	const int32_t win = (int32_t)std::min<int64_t>(prefetch_size, st->size - st->pos);
176	const int got = fetch_bytes(st->url, st->pos, win, st->cache.data());
177	if(got <= 0)
178	return got == 0 ? AVERROR_EOF : AVERROR(EIO);
179	st->cache_off = st->pos;
180	st->cache_len = got;
181
182	const int n = std::min(want, got);
183	std::memcpy(out, st->cache.data(), n);
184	st->pos += n;
185	return n;
186	}
187
188	static int64_t av_io_seek(void* opaque, int64_t offset, int whence)
189	{
190	auto* st = static_cast<AvStreamState*>(opaque);
191	switch(whence & ~AVSEEK_FORCE)
192	{
193	case AVSEEK_SIZE:
194	return st->size;
195	case SEEK_SET:
196	st->pos = offset;
197	break;
198	case SEEK_CUR:
199	st->pos += offset;
200	break;
201	case SEEK_END:
202	st->pos = st->size + offset;
203	break;
204	default:
205	return -1;
206	}
207	if(st->pos < 0)
208	st->pos = 0;
209	return st->pos;
210	}
211
212	#endif
213
NEW 214	AvIoDevice::AvIoDevice(const QString& path)	×
215	{
NEW 216	auto st = std::make_unique<AvStreamState>();	×
NEW 217	st->url = path.toStdString();	×
218
219	#if defined(__EMSCRIPTEN__)
220	bool ok = false;
221	const auto fetch_size = [&] {
222	qstdweb::File f = getWebFile(st->url);
223	if(f.file().isUndefined() \|\| f.file().isNull())
224	return;
225	st->size = (int64_t)f.size();
226	ok = st->size > 0;
227	};
228	if(emscripten_is_main_runtime_thread())
229	{
230	fetch_size();
231	}
232	else
233	{
234	emscripten_proxy_sync(
235	emscripten_proxy_get_system_queue(), emscripten_main_runtime_thread_id(),
236	[](void* p) { (static_cast<const decltype(fetch_size)>(p))(); },
237	const_cast<void>(static_cast<const void>(&fetch_size)));
238	}
239	if(!ok)
240	return;
241
242	st->cache.resize((size_t)std::min<int64_t>(prefetch_size, st->size));
243
244	auto* buffer = static_cast<unsigned char*>(av_malloc(av_io_buffer_size));
245	if(!buffer)
246	return;
247
248	avio = avio_alloc_context(
249	buffer, av_io_buffer_size, 0, st.get(), &av_io_read, nullptr, &av_io_seek);
250	if(!avio)
251	{
252	av_free(buffer);
253	return;
254	}
255	state = st.release();
256	#endif
NEW 257	}	×
258
NEW 259	AvIoDevice::~AvIoDevice()	×
260	{
NEW 261	if(avio)	×
262	{
NEW 263	av_freep(&avio->buffer);	×
NEW 264	avio_context_free(&avio);	×
NEW 265	}	×
NEW 266	delete state;	×
NEW 267	state = nullptr;	×
NEW 268	}	×
269
NEW 270	AvIoDevice::AvIoDevice(AvIoDevice&& other) noexcept	×
NEW 271	: avio{other.avio}	×
NEW 272	, state{other.state}	×
273	{
NEW 274	other.avio = nullptr;	×
NEW 275	other.state = nullptr;	×
NEW 276	}	×
277
NEW 278	AvIoDevice& AvIoDevice::operator=(AvIoDevice&& other) noexcept	×
279	{
NEW 280	if(this != &other)	×
281	{
NEW 282	if(avio)	×
283	{
NEW 284	av_freep(&avio->buffer);	×
NEW 285	avio_context_free(&avio);	×
NEW 286	}	×
NEW 287	delete state;	×
NEW 288	avio = other.avio;	×
NEW 289	state = other.state;	×
NEW 290	other.avio = nullptr;	×
NEW 291	other.state = nullptr;	×
NEW 292	}	×
NEW 293	return *this;	×
294	}
295
NEW 296	bool open_input_custom_io(	×
297	AVFormatContext*& format, AvIoDevice& io, const QString& path) noexcept
298	{
NEW 299	io = AvIoDevice{path};	×
NEW 300	if(!io)	×
NEW 301	return false;	×
302
NEW 303	AVFormatContext* fmt = avformat_alloc_context();	×
NEW 304	if(!fmt)	×
305	{
NEW 306	io = {};	×
NEW 307	return false;	×
308	}
309
NEW 310	fmt->pb = io.avio;	×
NEW 311	fmt->flags \|= AVFMT_FLAG_CUSTOM_IO;	×
312
NEW 313	if(avformat_open_input(&fmt, nullptr, nullptr, nullptr) != 0)	×
314	{
NEW 315	io = {};	×
NEW 316	return false;	×
317	}
318
NEW 319	format = fmt;	×
NEW 320	return true;	×
NEW 321	}	×
322
323	// libossia's libav_handle is Qt-agnostic; it delegates the browser-only Blob
324	// streaming back here through a hook so the submodule keeps no Qt dependency.
NEW 325	static AVFormatContext* libav_stream_open_hook(	×
326	const char* path, void& io_owner, void (&io_free)(void*)) noexcept
327	{
NEW 328	const QString qpath = QString::fromUtf8(path);	×
NEW 329	if(!isStreamedMediaPath(qpath))	×
NEW 330	return nullptr;	×
331
NEW 332	auto io = std::make_unique<AvIoDevice>();	×
NEW 333	AVFormatContext* fmt{};	×
NEW 334	if(!open_input_custom_io(fmt, *io, qpath))	×
NEW 335	return nullptr;	×
336
NEW 337	io_owner = io.release();	×
NEW 338	io_free = [](void* p) { delete static_cast<AvIoDevice*>(p); };	×
NEW 339	return fmt;	×
NEW 340	}	×
341
342	static const bool registered_libav_hook = [] {	64✔
343	ossia::libav_custom_open() = &libav_stream_open_hook;	32✔
344	return true;	32✔
345	}();
346	}
347	#endif

ossia / score / 30161608849

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous