Compare commits

...

31 Commits
v0.68 ... v0.76

Author SHA1 Message Date
Devaev Maxim
c94117ae1e Bump version: 0.75 → 0.76 2019-05-26 05:55:48 +03:00
Devaev Maxim
06a80df708 More flexible build 2019-05-26 05:55:35 +03:00
Devaev Maxim
13dff256c8 workers by number of cores 2019-05-25 00:52:49 +03:00
Devaev Maxim
07e9dbc0f7 fixed gcc options ordering 2019-05-24 20:13:09 +03:00
Devaev Maxim
c83dddbf0b fixed indent 2019-05-18 23:51:14 +03:00
Devaev Maxim
5672d1aa75 Bump version: 0.74 → 0.75 2019-05-18 18:24:32 +03:00
Devaev Maxim
078097efec Added links to FreeBSD port 2019-05-18 18:22:37 +03:00
Devaev Maxim
3cd8338886 Issue #6: make install-strip and make uninstall 2019-05-18 17:58:23 +03:00
Maxim Devaev
d2b57cc7d5 Update README.md 2019-05-18 17:25:42 +03:00
Maxim Devaev
2c4f59c87a Update README.ru.md 2019-05-18 17:24:53 +03:00
Devaev Maxim
36eb7eeb76 minor help fix 2019-05-18 16:35:10 +03:00
Devaev Maxim
3b6544db8a cpu/encoder: removed TODO about jpeg error handling 2019-05-18 02:12:18 +03:00
Devaev Maxim
8033ab23ed Bump version: 0.73 → 0.74 2019-05-17 17:15:17 +03:00
Devaev Maxim
ddb3db8b20 Issue #5: Added libevent version check 2019-05-17 16:57:53 +03:00
Devaev Maxim
b3c8071edb Bump version: 0.72 → 0.73 2019-05-09 18:51:27 +03:00
Devaev Maxim
bc70faae09 fix 2019-05-05 04:26:25 +03:00
Devaev Maxim
ae7c4c91e0 Bump version: 0.71 → 0.72 2019-05-03 01:49:44 +03:00
Devaev Maxim
613baa4e1e better unix socket handling 2019-05-03 00:10:48 +03:00
Devaev Maxim
33a101b4f7 Bump version: 0.70 → 0.71 2019-05-01 21:00:22 +03:00
Devaev Maxim
9e9039c4e6 y != yes 2019-05-01 20:59:30 +03:00
Maxim Devaev
b3ceec51de Update README.md 2019-04-26 02:24:16 +03:00
Maxim Devaev
e3ac4ba6f5 Update README.ru.md 2019-04-26 02:23:54 +03:00
Devaev Maxim
fa6b8b44c1 Bump version: 0.69 → 0.70 2019-04-18 05:17:00 +03:00
Devaev Maxim
8cb7574af2 encoder refactoring 2019-04-18 05:09:32 +03:00
Devaev Maxim
c34d644c2a Bump version: 0.68 → 0.69 2019-04-17 07:38:37 +03:00
Devaev Maxim
7857fa8f63 blank: handling libjpeg errors 2019-04-17 07:14:19 +03:00
Devaev Maxim
3253de83dc refactoring 2019-04-14 06:09:21 +03:00
Devaev Maxim
030464c3b8 refactoring 2019-04-14 03:41:35 +03:00
Devaev Maxim
0020aa69ec configurable blank page 2019-04-14 03:35:25 +03:00
Devaev Maxim
93bfa56ccf unicode fix 2019-04-13 23:02:26 +03:00
Devaev Maxim
3392ac5fbc refactoring 2019-04-13 23:02:12 +03:00
24 changed files with 427 additions and 167 deletions

View File

@@ -1,7 +1,7 @@
[bumpversion]
commit = True
tag = True
current_version = 0.68
current_version = 0.76
parse = (?P<major>\d+)\.(?P<minor>\d+)(\.(?P<patch>\d+)(\-(?P<release>[a-z]+))?)?
serialize =
{major}.{minor}

View File

@@ -1,24 +1,27 @@
PROG ?= ustreamer
DESTDIR ?=
PREFIX ?= /usr/local
CC ?= gcc
CFLAGS ?= -O3
LDFLAGS ?=
CC ?= gcc
RPI_VC_HEADERS ?= /opt/vc/include
RPI_VC_LIBS ?= /opt/vc/lib
# =====
LIBS = -lm -ljpeg -pthread -levent -levent_pthreads -luuid
override CFLAGS += -c -std=c11 -Wall -Wextra -D_GNU_SOURCE
SOURCES = $(shell ls src/*.c src/http/*.c src/encoders/cpu/*.c src/encoders/hw/*.c)
PROG = ustreamer
ifeq ($(shell ls -d /opt/vc/include 2>/dev/null), /opt/vc/include)
SOURCES += $(shell ls src/encoders/omx/*.c)
LIBS += -lbcm_host -lvcos -lopenmaxil -L/opt/vc/lib
override CFLAGS += -DWITH_OMX_ENCODER -DOMX_SKIP64BIT -I/opt/vc/include
ifeq ($(WITH_OMX_ENCODER),)
else
LIBS += -lbcm_host -lvcos -lopenmaxil -L$(RPI_VC_LIBS)
override CFLAGS += -DWITH_OMX_ENCODER -DOMX_SKIP64BIT -I$(RPI_VC_HEADERS)
SOURCES += $(shell ls src/encoders/omx/*.c)
endif
OBJECTS = $(SOURCES:.c=.o)
# =====
all: $(SOURCES) $(PROG)
@@ -28,17 +31,32 @@ install: $(PROG)
install -Dm755 $(PROG) $(DESTDIR)$(PREFIX)/bin/$(PROG)
install-strip: install
strip $(DESTDIR)$(PREFIX)/bin/$(PROG)
uninstall:
rm $(DESTDIR)$(PREFIX)/bin/$(PROG)
regen:
tools/make-jpeg-h.py src/http/data/blank.jpeg src/http/data/blank_jpeg.h BLANK
tools/make-html-h.py src/http/data/index.html src/http/data/index_html.h INDEX
$(PROG): $(OBJECTS)
$(CC) $(LIBS) $(LDFLAGS) $(OBJECTS) -o $@
$(PROG): $(SOURCES:.c=.o)
$(info -- LINKING $@)
@ $(CC) $(SOURCES:.c=.o) -o $@ $(LDFLAGS) $(LIBS)
$(info ===== Build complete =====)
$(info == CC = $(CC))
$(info == LIBS = $(LIBS))
$(info == CFLAGS = $(CFLAGS))
$(info == LDFLAGS = $(LDFLAGS))
.c.o:
$(CC) $(LIBS) $(CFLAGS) $< -o $@
$(info -- CC $<)
@ $(CC) $< -o $@ $(CFLAGS) $(LIBS)
release:
@@ -50,13 +68,14 @@ release:
bump:
bumpversion $(if $(V), $(V), minor)
bumpversion $(if $(V),$(V),minor)
push:
git push
git push --tags
clean-all: clean
clean:
find src -name '*.o' -exec rm '{}' \;

View File

@@ -31,7 +31,7 @@ If you're going to live-stream from your backyard webcam and need to control it,
# Building
You'll need ```make```, ```gcc```, ```libevent``` with ```pthreads``` support, ```libjpeg8```/```libjpeg-turbo``` and ```libuuid```.
It should compile automatically with OpenMAX IL on Raspberry Pi, if the corresponding headers are present in ```/opt/vc/include```.
On Raspberry Pi you can build the program with OpenMAX IL. To do this pass option ```WITH_OMX_ENCODER=1``` to ```make```.
```
$ git clone --depth=1 https://github.com/pi-kvm/ustreamer
@@ -40,7 +40,8 @@ $ make
$ ./ustreamer --help
```
AUR has a package for Arch Linux: https://aur.archlinux.org/packages/ustreamer
AUR has a package for Arch Linux: https://aur.archlinux.org/packages/ustreamer. It should compile automatically with OpenMAX IL on Raspberry Pi, if the corresponding headers are present in ```/opt/vc/include```.
FreeBSD port: https://www.freshports.org/multimedia/ustreamer.
-----
# Usage

View File

@@ -31,7 +31,7 @@
# Сборка
Для сборки вам понадобятся ```make```, ```gcc```, ```libevent``` с поддержкой ```pthreads```, ```libjpeg8```/```libjpeg-turbo``` и ```libuuid```.
На Raspberry Pi програма автоматически собирается с поддержкой OpenMAX IL, если обнаружит нужные хедеры в ```/opt/vc/include```.
На Raspberry Pi программу можно собрать с поддержкой OpenMAX IL. Для этого передайте ```make``` параметр ```WITH_OMX_ENCODER=1```.
```
$ git clone --depth=1 https://github.com/pi-kvm/ustreamer
@@ -40,7 +40,8 @@ $ make
$ ./ustreamer --help
```
Для Arch Linux в AUR есть готовый пакет: https://aur.archlinux.org/packages/ustreamer
Для Arch Linux в AUR есть готовый пакет: https://aur.archlinux.org/packages/ustreamer. На Raspberry Pi програма автоматически собирается с поддержкой OpenMAX IL, если обнаружит нужные хедеры в ```/opt/vc/include```.
Порт для FreeBSD: https://www.freshports.org/multimedia/ustreamer.
-----
# Использование

View File

@@ -3,7 +3,7 @@
pkgname=ustreamer
pkgver=0.68
pkgver=0.76
pkgrel=1
pkgdesc="Lightweight and fast MJPG-HTTP streamer"
url="https://github.com/pi-kvm/ustreamer"
@@ -12,19 +12,23 @@ arch=(i686 x86_64 armv6h armv7h)
depends=(libjpeg libevent libutil-linux)
# optional: raspberrypi-firmware for OMX JPEG encoder
makedepends=(gcc make)
source=("$url/archive/v$pkgver.tar.gz")
source=(${pkgname}::"git+https://github.com/pi-kvm/ustreamer#commit=${pkgver}")
md5sums=(SKIP)
build() {
cd $srcdir
cd "$srcdir"
rm -rf $pkgname-build
cp -r ustreamer-$pkgver $pkgname-build
cp -r $pkgname $pkgname-build
cd $pkgname-build
make CFLAGS="$CFLAGS" LDFLAGS="$LDFLAGS" $MAKEFLAGS
local _options=""
[ -d /opt/vc/include ] && _options="$_options WITH_OMX_ENCODER=1"
make $_options CFLAGS="$CFLAGS" LDFLAGS="$LDFLAGS" $MAKEFLAGS
}
package() {
cd $srcdir/$pkgname-build
cd "$srcdir/$pkgname-build"
make DESTDIR="$pkgdir" PREFIX=/usr install
}

View File

@@ -5,7 +5,7 @@ EAPI=7
inherit git-r3
DESCRIPTION="µStreamer - Lightweight and fast MJPG-HTTP streamer"
DESCRIPTION="uStreamer - Lightweight and fast MJPG-HTTP streamer"
HOMEPAGE="https://github.com/pi-kvm/ustreamer"
EGIT_REPO_URI="https://github.com/pi-kvm/ustreamer.git"

View File

@@ -6,7 +6,7 @@
include $(TOPDIR)/rules.mk
PKG_NAME:=ustreamer
PKG_VERSION:=0.68
PKG_VERSION:=0.76
PKG_RELEASE:=1
PKG_MAINTAINER:=Maxim Devaev <mdevaev@gmail.com>

View File

@@ -22,4 +22,6 @@
#pragma once
#define VERSION "0.68"
#ifndef VERSION
# define VERSION "0.76"
#endif

View File

@@ -85,6 +85,12 @@ struct device_t *device_init() {
struct controls_t *ctl;
struct device_runtime_t *run;
struct device_t *dev;
long cores_sysconf;
unsigned cores_available;
cores_sysconf = sysconf(_SC_NPROCESSORS_ONLN);
cores_sysconf = (cores_sysconf < 0 ? 0 : cores_sysconf);
cores_available = max_u(min_u(cores_sysconf, 4), 1);
A_CALLOC(ctl, 1);
@@ -97,8 +103,8 @@ struct device_t *device_init() {
dev->height = 480;
dev->format = V4L2_PIX_FMT_YUYV;
dev->standard = V4L2_STD_UNKNOWN;
dev->n_buffers = max_u(min_u(sysconf(_SC_NPROCESSORS_ONLN), 4), 1) + 1;
dev->n_workers = dev->n_buffers;
dev->n_buffers = cores_available + 1;
dev->n_workers = min_u(cores_available, dev->n_buffers);
dev->timeout = 1;
dev->error_delay = 1;
dev->ctl = ctl;
@@ -156,6 +162,8 @@ int device_open(struct device_t *dev) {
_device_open_alloc_picbufs(dev);
_device_apply_controls(dev);
dev->run->n_workers = min_u(dev->run->n_buffers, dev->n_workers);
LOG_DEBUG("Device fd=%d initialized", dev->run->fd);
return 0;
@@ -165,6 +173,8 @@ int device_open(struct device_t *dev) {
}
void device_close(struct device_t *dev) {
dev->run->n_workers = 0;
if (dev->run->pictures) {
LOG_DEBUG("Releasing picture buffers ...");
for (unsigned index = 0; index < dev->run->n_buffers && dev->run->pictures[index].data; ++index) {
@@ -318,7 +328,7 @@ static int _device_open_check_cap(struct device_t *dev) {
static int _device_open_dv_timings(struct device_t *dev) {
_device_apply_resolution(dev, dev->width, dev->height);
if (dev->dv_timings) {
LOG_DEBUG("Using DV-timings");
LOG_DEBUG("Using DV timings");
if (_device_apply_dv_timings(dev) < 0) {
return -1;

View File

@@ -63,7 +63,7 @@ struct device_runtime_t {
unsigned height;
unsigned format;
unsigned n_buffers;
// unsigned n_workers; // FIXME
unsigned n_workers;
struct hw_buffer_t *hw_buffers;
struct picture_t *pictures;
size_t max_raw_image_size;

View File

@@ -20,6 +20,7 @@
*****************************************************************************/
#include <stdbool.h>
#include <strings.h>
#include <assert.h>
@@ -66,48 +67,6 @@ struct encoder_t *encoder_init() {
return encoder;
}
#pragma GCC diagnostic ignored "-Wunused-parameter"
#pragma GCC diagnostic push
void encoder_prepare(struct encoder_t *encoder, struct device_t *dev) {
#pragma GCC diagnostic pop
assert(encoder->type != ENCODER_TYPE_UNKNOWN);
// XXX: Тут нет гонки, потому что encoder_prepare() запускается еще до существования других потоков
encoder->run->type = encoder->type;
encoder->run->quality = encoder->quality;
LOG_INFO("Using JPEG quality: %u%%", encoder->quality);
# ifdef WITH_OMX_ENCODER
if (encoder->run->type == ENCODER_TYPE_OMX) {
LOG_DEBUG("Preparing OMX JPEG encoder ...");
if (dev->n_workers > OMX_MAX_ENCODERS) {
LOG_INFO("OMX JPEG encoder sets limit for worker threads: %u", OMX_MAX_ENCODERS);
dev->n_workers = OMX_MAX_ENCODERS;
}
encoder->run->n_omxs = dev->n_workers;
A_CALLOC(encoder->run->omxs, encoder->run->n_omxs);
for (unsigned index = 0; index < encoder->run->n_omxs; ++index) {
if ((encoder->run->omxs[index] = omx_encoder_init()) == NULL) {
goto use_fallback;
}
}
}
# endif
return;
# pragma GCC diagnostic ignored "-Wunused-label"
# pragma GCC diagnostic push
use_fallback:
LOG_ERROR("Can't initialize selected encoder, using CPU instead it");
encoder->run->type = ENCODER_TYPE_CPU;
encoder->run->quality = encoder->quality;
# pragma GCC diagnostic pop
}
void encoder_destroy(struct encoder_t *encoder) {
# ifdef WITH_OMX_ENCODER
if (encoder->run->omxs) {
@@ -142,52 +101,87 @@ const char *encoder_type_to_string(enum encoder_type_t type) {
return _ENCODER_TYPES[0].name;
}
void encoder_prepare_live(struct encoder_t *encoder, struct device_t *dev) {
assert(encoder->run->type != ENCODER_TYPE_UNKNOWN);
void encoder_prepare(struct encoder_t *encoder, struct device_t *dev) {
enum encoder_type_t type = (encoder->run->cpu_forced ? ENCODER_TYPE_CPU : encoder->type);
unsigned quality = encoder->quality;
bool cpu_forced = false;
if (
(dev->run->format == V4L2_PIX_FMT_MJPEG || dev->run->format == V4L2_PIX_FMT_JPEG)
&& encoder->run->type != ENCODER_TYPE_HW
) {
if ((dev->run->format == V4L2_PIX_FMT_MJPEG || dev->run->format == V4L2_PIX_FMT_JPEG) && type != ENCODER_TYPE_HW) {
LOG_INFO("Switching to HW JPEG encoder because the input format is (M)JPEG");
A_MUTEX_LOCK(&encoder->run->mutex);
encoder->run->type = ENCODER_TYPE_HW;
A_MUTEX_UNLOCK(&encoder->run->mutex);
type = ENCODER_TYPE_HW;
}
if (encoder->run->type == ENCODER_TYPE_HW) {
if (type == ENCODER_TYPE_HW) {
if (dev->run->format != V4L2_PIX_FMT_MJPEG && dev->run->format != V4L2_PIX_FMT_JPEG) {
LOG_INFO("Switching to CPU JPEG encoder because the input format is not (M)JPEG");
goto use_fallback;
goto use_cpu;
}
if (hw_encoder_prepare_live(dev, encoder->quality) < 0) {
A_MUTEX_LOCK(&encoder->run->mutex);
encoder->run->quality = 0;
A_MUTEX_UNLOCK(&encoder->run->mutex);
LOG_INFO("Using JPEG quality: HW-default");
if (hw_encoder_prepare(dev, quality) < 0) {
quality = 0;
}
dev->run->n_workers = 1;
}
# ifdef WITH_OMX_ENCODER
else if (encoder->run->type == ENCODER_TYPE_OMX) {
else if (type == ENCODER_TYPE_OMX) {
LOG_DEBUG("Preparing OMX JPEG encoder ...");
if (dev->run->n_workers > OMX_MAX_ENCODERS) {
LOG_INFO("OMX JPEG encoder sets limit for worker threads: %u", OMX_MAX_ENCODERS);
dev->run->n_workers = OMX_MAX_ENCODERS;
}
if (encoder->run->omxs == NULL) {
A_CALLOC(encoder->run->omxs, OMX_MAX_ENCODERS);
}
// Начинаем с нуля и доинициализируем на следующих заходах при необходимости
for (; encoder->run->n_omxs < dev->run->n_workers; ++encoder->run->n_omxs) {
if ((encoder->run->omxs[encoder->run->n_omxs] = omx_encoder_init()) == NULL) {
LOG_ERROR("Can't initialize OMX JPEG encoder, falling back to CPU");
goto force_cpu;
}
}
for (unsigned index = 0; index < encoder->run->n_omxs; ++index) {
if (omx_encoder_prepare_live(encoder->run->omxs[index], dev, encoder->quality) < 0) {
if (omx_encoder_prepare(encoder->run->omxs[index], dev, quality) < 0) {
LOG_ERROR("Can't prepare OMX JPEG encoder, falling back to CPU");
goto use_fallback;
goto force_cpu;
}
}
}
# endif
return;
goto ok;
# pragma GCC diagnostic ignored "-Wunused-label"
# pragma GCC diagnostic push
force_cpu:
cpu_forced = true;
# pragma GCC diagnostic pop
use_cpu:
type = ENCODER_TYPE_CPU;
quality = encoder->quality;
ok:
if (quality == 0) {
LOG_INFO("Using JPEG quality: encoder default");
} else {
LOG_INFO("Using JPEG quality: %u%%", quality);
}
use_fallback:
A_MUTEX_LOCK(&encoder->run->mutex);
encoder->run->type = ENCODER_TYPE_CPU;
encoder->run->quality = encoder->quality;
encoder->run->type = type;
encoder->run->quality = quality;
if (cpu_forced) {
encoder->run->cpu_forced = true;
}
A_MUTEX_UNLOCK(&encoder->run->mutex);
}
#pragma GCC diagnostic ignored "-Wunused-label"
#pragma GCC diagnostic ignored "-Wunused-parameter"
#pragma GCC diagnostic push
int encoder_compress_buffer(struct encoder_t *encoder, struct device_t *dev, unsigned worker_number, unsigned buf_index) {
#pragma GCC diagnostic pop
@@ -195,14 +189,14 @@ int encoder_compress_buffer(struct encoder_t *encoder, struct device_t *dev, uns
assert(encoder->run->type != ENCODER_TYPE_UNKNOWN);
if (encoder->run->type == ENCODER_TYPE_CPU) {
cpu_encoder_compress_buffer(dev, buf_index, encoder->quality);
cpu_encoder_compress_buffer(dev, buf_index, encoder->run->quality);
} else if (encoder->run->type == ENCODER_TYPE_HW) {
hw_encoder_compress_buffer(dev, buf_index);
}
# ifdef WITH_OMX_ENCODER
else if (encoder->run->type == ENCODER_TYPE_OMX) {
if (omx_encoder_compress_buffer(encoder->run->omxs[worker_number], dev, buf_index) < 0) {
goto use_fallback;
goto error;
}
}
# endif
@@ -211,11 +205,10 @@ int encoder_compress_buffer(struct encoder_t *encoder, struct device_t *dev, uns
# pragma GCC diagnostic ignored "-Wunused-label"
# pragma GCC diagnostic push
use_fallback:
LOG_INFO("Error while compressing, falling back to CPU");
error:
LOG_INFO("Error while compressing buffer, falling back to CPU");
A_MUTEX_LOCK(&encoder->run->mutex);
encoder->run->type = ENCODER_TYPE_CPU;
encoder->run->quality = encoder->quality;
encoder->run->cpu_forced = true;
A_MUTEX_UNLOCK(&encoder->run->mutex);
return -1;
# pragma GCC diagnostic pop

View File

@@ -22,6 +22,8 @@
#pragma once
#include <stdbool.h>
#include "pthread.h"
#include "tools.h"
@@ -51,6 +53,7 @@ enum encoder_type_t {
struct encoder_runtime_t {
enum encoder_type_t type;
unsigned quality;
bool cpu_forced;
pthread_mutex_t mutex;
# ifdef WITH_OMX_ENCODER
@@ -74,6 +77,4 @@ enum encoder_type_t encoder_parse_type(const char *str);
const char *encoder_type_to_string(enum encoder_type_t type);
void encoder_prepare(struct encoder_t *encoder, struct device_t *dev);
void encoder_prepare_live(struct encoder_t *encoder, struct device_t *dev);
int encoder_compress_buffer(struct encoder_t *encoder, struct device_t *dev, unsigned worker_number, unsigned buf_index);

View File

@@ -105,8 +105,6 @@ void cpu_encoder_compress_buffer(struct device_t *dev, unsigned index, unsigned
# undef WRITE_SCANLINES
// TODO: process jpeg errors:
// https://stackoverflow.com/questions/19857766/error-handling-in-libjpeg
jpeg_finish_compress(&jpeg);
jpeg_destroy_compress(&jpeg);

View File

@@ -44,7 +44,7 @@ static bool _is_huffman(const unsigned char *data);
static size_t _memcpy_with_huffman(unsigned char *dest, const unsigned char *src, size_t size);
int hw_encoder_prepare_live(struct device_t *dev, unsigned quality) {
int hw_encoder_prepare(struct device_t *dev, unsigned quality) {
struct v4l2_jpegcompression comp;
MEMSET_ZERO(comp);

View File

@@ -25,5 +25,5 @@
#include "../../device.h"
int hw_encoder_prepare_live(struct device_t *dev, unsigned quality);
int hw_encoder_prepare(struct device_t *dev, unsigned quality);
void hw_encoder_compress_buffer(struct device_t *dev, unsigned index);

View File

@@ -153,7 +153,7 @@ void omx_encoder_destroy(struct omx_encoder_t *omx) {
free(omx);
}
int omx_encoder_prepare_live(struct omx_encoder_t *omx, struct device_t *dev, unsigned quality) {
int omx_encoder_prepare(struct omx_encoder_t *omx, struct device_t *dev, unsigned quality) {
if (component_set_state(&omx->encoder, OMX_StateIdle) < 0) {
return -1;
}

View File

@@ -53,5 +53,5 @@ struct omx_encoder_t {
struct omx_encoder_t *omx_encoder_init();
void omx_encoder_destroy(struct omx_encoder_t *omx);
int omx_encoder_prepare_live(struct omx_encoder_t *omx, struct device_t *dev, unsigned quality);
int omx_encoder_prepare(struct omx_encoder_t *omx, struct device_t *dev, unsigned quality);
int omx_encoder_compress_buffer(struct omx_encoder_t *omx, struct device_t *dev, unsigned index);

174
src/http/blank.c Normal file
View File

@@ -0,0 +1,174 @@
/*****************************************************************************
# #
# uStreamer - Lightweight and fast MJPG-HTTP streamer. #
# #
# Copyright (C) 2018 Maxim Devaev <mdevaev@gmail.com> #
# #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
# #
*****************************************************************************/
#include <stdio.h>
#include <stdbool.h>
#include <string.h>
#include <setjmp.h>
#include <jpeglib.h>
#include "../tools.h"
#include "../logging.h"
#include "data/blank_jpeg.h"
#include "blank.h"
struct _jpeg_error_manager_t {
struct jpeg_error_mgr mgr; // Default manager
jmp_buf jmp;
};
static struct blank_t *_blank_init_internal();
static struct blank_t *_blank_init_external(const char *path);
static int _jpeg_read_geometry(FILE *fp, unsigned *width, unsigned *height);
static void _jpeg_error_handler(j_common_ptr jpeg);
struct blank_t *blank_init(const char *path) {
struct blank_t *blank = NULL;
if (path) {
blank = _blank_init_external(path);
}
if (blank) {
LOG_INFO("Using external blank placeholder: %s", path);
} else {
blank = _blank_init_internal();
LOG_INFO("Using internal blank placeholder");
}
return blank;
}
void blank_destroy(struct blank_t *blank) {
free(blank->picture.data);
free(blank);
}
static struct blank_t *_blank_init_internal() {
struct blank_t *blank;
A_CALLOC(blank, 1);
A_CALLOC(blank->picture.data, ARRAY_LEN(BLANK_JPEG_DATA));
memcpy(blank->picture.data, BLANK_JPEG_DATA, ARRAY_LEN(BLANK_JPEG_DATA));
blank->picture.used = ARRAY_LEN(BLANK_JPEG_DATA);
blank->picture.allocated = ARRAY_LEN(BLANK_JPEG_DATA);
blank->width = BLANK_JPEG_WIDTH;
blank->height = BLANK_JPEG_HEIGHT;
return blank;
}
static struct blank_t *_blank_init_external(const char *path) {
FILE *fp = NULL;
struct blank_t *blank;
A_CALLOC(blank, 1);
if ((fp = fopen(path, "rb")) == NULL) {
LOG_PERROR("Can't open blank placeholder '%s'", path);
goto error;
}
if (_jpeg_read_geometry(fp, &blank->width, &blank->height) < 0) {
goto error;
}
if (fseek(fp, 0, SEEK_SET) < 0) {
LOG_PERROR("Can't seek to begin of the blank placeholder");
goto error;
}
# define CHUNK_SIZE (100 * 1024)
while (true) {
if (blank->picture.used + CHUNK_SIZE >= blank->picture.allocated) {
blank->picture.allocated = blank->picture.used + CHUNK_SIZE * 2;
A_REALLOC(blank->picture.data, blank->picture.allocated);
}
size_t readed = fread(blank->picture.data + blank->picture.used, 1, CHUNK_SIZE, fp);
blank->picture.used += readed;
if (readed < CHUNK_SIZE) {
if (feof(fp)) {
goto ok;
} else {
LOG_PERROR("Can't read blank placeholder");
goto error;
}
}
}
# undef CHUNK_SIZE
error:
free(blank->picture.data);
free(blank);
blank = NULL;
ok:
if (fp) {
fclose(fp);
}
return blank;
}
static int _jpeg_read_geometry(FILE *fp, unsigned *width, unsigned *height) {
struct jpeg_decompress_struct jpeg;
struct _jpeg_error_manager_t jpeg_error;
jpeg_create_decompress(&jpeg);
// https://stackoverflow.com/questions/19857766/error-handling-in-libjpeg
jpeg.err = jpeg_std_error((struct jpeg_error_mgr *)&jpeg_error);
jpeg_error.mgr.error_exit = _jpeg_error_handler;
if (setjmp(jpeg_error.jmp) < 0) {
jpeg_destroy_decompress(&jpeg);
return -1;
}
jpeg_stdio_src(&jpeg, fp);
jpeg_read_header(&jpeg, TRUE);
jpeg_start_decompress(&jpeg);
*width = jpeg.output_width;
*height = jpeg.output_height;
jpeg_destroy_decompress(&jpeg);
return 0;
}
static void _jpeg_error_handler(j_common_ptr jpeg) {
struct _jpeg_error_manager_t *jpeg_error = (struct _jpeg_error_manager_t *)jpeg->err;
char msg[JMSG_LENGTH_MAX];
(*jpeg_error->mgr.format_message)(jpeg, msg);
LOG_ERROR("Invalid blank placeholder: %s", msg);
longjmp(jpeg_error->jmp, -1);
}

38
src/http/blank.h Normal file
View File

@@ -0,0 +1,38 @@
/*****************************************************************************
# #
# uStreamer - Lightweight and fast MJPG-HTTP streamer. #
# #
# Copyright (C) 2018 Maxim Devaev <mdevaev@gmail.com> #
# #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
# #
*****************************************************************************/
#pragma once
#include <stdlib.h>
#include "../device.h"
struct blank_t {
struct picture_t picture;
unsigned width;
unsigned height;
};
struct blank_t *blank_init(const char *path);
void blank_destroy(struct blank_t *blank);

View File

@@ -52,13 +52,13 @@
#include "../encoder.h"
#include "../stream.h"
#include "blank.h"
#include "base64.h"
#include "mime.h"
#include "static.h"
#include "server.h"
#include "data/index_html.h"
#include "data/blank_jpeg.h"
static bool _http_get_param_true(struct evkeyvalq *params, const char *key);
@@ -103,8 +103,6 @@ struct http_server_t *http_server_init(struct stream_t *stream) {
server->timeout = 10;
server->run = run;
_expose_blank_picture(server);
assert(!evthread_use_pthreads());
assert((run->base = event_base_new()));
assert((run->http = evhttp_new(run->base)));
@@ -123,7 +121,10 @@ void http_server_destroy(struct http_server_t *server) {
close(server->run->unix_fd);
}
event_base_free(server->run->base);
# if LIBEVENT_VERSION_NUMBER >= 0x02010100
libevent_global_shutdown();
# endif
for (struct stream_client_t *client = server->run->stream_clients; client != NULL;) {
struct stream_client_t *next = client->next;
@@ -137,6 +138,10 @@ void http_server_destroy(struct http_server_t *server) {
free(server->run->auth_token);
}
if (server->run->blank) {
blank_destroy(server->run->blank);
}
free(server->run->exposed->picture.data);
free(server->run->exposed);
free(server->run);
@@ -155,6 +160,10 @@ int http_server_listen(struct http_server_t *server) {
assert(!evhttp_set_cb(server->run->http, "/stream", _http_callback_stream, (void *)server));
}
server->run->drop_same_frames_blank = max_u(server->drop_same_frames, server->run->drop_same_frames_blank);
server->run->blank = blank_init(server->blank_path);
_expose_blank_picture(server);
{
struct timeval refresh_interval;
@@ -169,8 +178,6 @@ int http_server_listen(struct http_server_t *server) {
assert(!event_add(server->run->refresh, &refresh_interval));
}
server->run->drop_same_frames_blank = max_u(server->drop_same_frames, server->run->drop_same_frames_blank);
if (server->slowdown) {
stream_switch_slowdown(server->run->stream, true);
}
@@ -195,19 +202,25 @@ int http_server_listen(struct http_server_t *server) {
if (server->unix_path[0] != '\0') {
struct sockaddr_un unix_addr;
int unix_fd_flags;
LOG_DEBUG("Binding HTTP to UNIX socket '%s' ...", server->unix_path);
assert((server->run->unix_fd = socket(AF_UNIX, SOCK_STREAM, 0)));
assert((unix_fd_flags = fcntl(server->run->unix_fd, F_GETFL)) >= 0);
unix_fd_flags |= O_NONBLOCK;
assert(fcntl(server->run->unix_fd, F_SETFL, unix_fd_flags) >= 0);
# define MAX_SUN_PATH (sizeof(unix_addr.sun_path) - 1)
strncpy(unix_addr.sun_path, server->unix_path, 107);
unix_addr.sun_path[107] = '\0';
if (strlen(server->unix_path) > MAX_SUN_PATH) {
LOG_ERROR("UNIX socket path is too long, max=%zu", MAX_SUN_PATH);
return -1;
}
MEMSET_ZERO(unix_addr);
strncpy(unix_addr.sun_path, server->unix_path, MAX_SUN_PATH);
unix_addr.sun_family = AF_UNIX;
# undef MAX_SUN_PATH
assert((server->run->unix_fd = socket(AF_UNIX, SOCK_STREAM, 0)) >= 0);
assert(!evutil_make_socket_nonblocking(server->run->unix_fd));
if (server->unix_rm && unlink(server->unix_path) < 0) {
if (errno != ENOENT) {
LOG_PERROR("Can't remove old UNIX socket '%s'", server->unix_path);
@@ -262,7 +275,6 @@ static bool _http_get_param_true(struct evkeyvalq *params, const char *key) {
if ((value_str = evhttp_find_header(params, key)) != NULL) {
if (
value_str[0] == '1'
|| value_str[0] == 'y'
|| !evutil_ascii_strcasecmp(value_str, "true")
|| !evutil_ascii_strcasecmp(value_str, "yes")
) {
@@ -523,7 +535,7 @@ static void _http_callback_stream(struct evhttp_request *request, void *v_server
PREPROCESS_REQUEST;
conn = evhttp_request_get_connection(request);
if (conn != NULL) {
if (conn) {
A_CALLOC(client, 1);
client->server = server;
client->request = request;
@@ -641,8 +653,9 @@ static void _http_callback_stream_write(struct bufferevent *buf_event, void *v_c
"Content-Length: %zu" RN
"X-Timestamp: %.06Lf" RN
"%s",
EXPOSED(picture.used) * sizeof(*EXPOSED(picture.data)),
get_now_real(), (client->extra_headers ? "" : RN)
EXPOSED(picture.used),
get_now_real(),
(client->extra_headers ? "" : RN)
));
if (client->extra_headers) {
assert(evbuffer_add_printf(buf,
@@ -675,10 +688,7 @@ static void _http_callback_stream_write(struct bufferevent *buf_event, void *v_c
}
}
assert(!evbuffer_add(buf,
(void *)EXPOSED(picture.data),
EXPOSED(picture.used) * sizeof(*EXPOSED(picture.data))
));
assert(!evbuffer_add(buf, (void *)EXPOSED(picture.data), EXPOSED(picture.used)));
assert(evbuffer_add_printf(buf, RN "--" BOUNDARY RN));
if (client->advance_headers) {
@@ -711,12 +721,12 @@ static void _http_callback_stream_error(UNUSED struct bufferevent *buf_event, UN
}
conn = evhttp_request_get_connection(client->request);
if (conn != NULL) {
if (conn) {
evhttp_connection_get_peer(conn, &client_addr, &client_port);
}
LOG_INFO("HTTP: Disconnected the stream client: [%s]:%u; clients now: %u",
client_addr, client_port, RUN(stream_clients_count));
if (conn != NULL) {
if (conn) {
evhttp_connection_free(conn);
}
@@ -744,7 +754,7 @@ static void _http_queue_send_stream(struct http_server_t *server, bool stream_up
for (struct stream_client_t *client = server->run->stream_clients; client != NULL; client = client->next) {
conn = evhttp_request_get_connection(client->request);
if (conn != NULL) {
if (conn) {
// Фикс для бага WebKit. При включенной опции дропа одинаковых фреймов,
// WebKit отрисовывает последний фрейм в серии с некоторой задержкой,
// и нужно послать два фрейма, чтобы серия была вовремя завершена.
@@ -824,8 +834,7 @@ static bool _expose_new_picture(struct http_server_t *server) {
EXPOSED(expose_begin_time) = get_now_monotonic();
# define MEM_STREAM_TO_EXPOSED \
EXPOSED(picture.data), STREAM(picture.data), \
STREAM(picture.used) * sizeof(*STREAM(picture.data))
EXPOSED(picture.data), STREAM(picture.data), STREAM(picture.used)
if (server->drop_same_frames) {
if (
@@ -878,28 +887,28 @@ static bool _expose_new_picture(struct http_server_t *server) {
}
static bool _expose_blank_picture(struct http_server_t *server) {
# define BLANK(_next) server->run->blank->_next
# define EXPOSED(_next) server->run->exposed->_next
# define BLANK_JPEG_LEN ARRAY_LEN(BLANK_JPEG_DATA)
EXPOSED(expose_begin_time) = get_now_monotonic();
EXPOSED(expose_cmp_time) = EXPOSED(expose_begin_time);
if (EXPOSED(online) || EXPOSED(picture.used) == 0) {
if (EXPOSED(picture.allocated) < BLANK_JPEG_LEN) {
A_REALLOC(EXPOSED(picture.data), BLANK_JPEG_LEN);
EXPOSED(picture.allocated) = BLANK_JPEG_LEN;
if (EXPOSED(picture.allocated) < BLANK(picture.used)) {
A_REALLOC(EXPOSED(picture.data), BLANK(picture.used));
EXPOSED(picture.allocated) = BLANK(picture.used);
}
memcpy(EXPOSED(picture.data), BLANK_JPEG_DATA, BLANK_JPEG_LEN * sizeof(*EXPOSED(picture.data)));
memcpy(EXPOSED(picture.data), BLANK(picture.data), BLANK(picture.used));
EXPOSED(picture.used) = BLANK_JPEG_LEN;
EXPOSED(picture.used) = BLANK(picture.used);
EXPOSED(picture.grab_time) = 0;
EXPOSED(picture.encode_begin_time) = 0;
EXPOSED(picture.encode_end_time) = 0;
EXPOSED(width) = BLANK_JPEG_WIDTH;
EXPOSED(height) = BLANK_JPEG_HEIGHT;
EXPOSED(width) = BLANK(width);
EXPOSED(height) = BLANK(height);
EXPOSED(captured_fps) = 0;
EXPOSED(online) = false;
goto updated;
@@ -917,6 +926,6 @@ static bool _expose_blank_picture(struct http_server_t *server) {
EXPOSED(expose_end_time) = get_now_monotonic();
return true; // Updated
# undef BLANK_JPEG_LEN
# undef EXPOSED
# undef BLANK
}

View File

@@ -33,6 +33,8 @@
#include "../tools.h"
#include "../stream.h"
#include "blank.h"
struct stream_client_t {
struct http_server_t *server;
@@ -78,6 +80,7 @@ struct http_server_runtime_t {
struct exposed_t *exposed;
struct stream_client_t *stream_clients;
unsigned stream_clients_count;
struct blank_t *blank;
unsigned drop_same_frames_blank;
};
@@ -87,14 +90,17 @@ struct http_server_t {
char *unix_path;
bool unix_rm;
mode_t unix_mode;
unsigned timeout;
char *user;
char *passwd;
char *static_path;
char *blank_path;
unsigned drop_same_frames;
bool slowdown;
unsigned fake_width;
unsigned fake_height;
unsigned timeout;
struct http_server_runtime_t *run;
};

View File

@@ -123,10 +123,10 @@ pthread_mutex_t log_mutex;
INLINE char *errno_to_string(char *buf, size_t size) {
#if defined(__GLIBC__) && defined(_GNU_SOURCE)
# if defined(__GLIBC__) && defined(_GNU_SOURCE)
return strerror_r(errno, buf, size);
#else
# else
strerror_r(errno, buf, size);
return buf;
#endif
# endif
}

View File

@@ -41,7 +41,7 @@
#include "http/server.h"
static const char _SHORT_OPTS[] = "d:i:x:y:m:a:f:z:ntb:w:q:c:s:p:u:ro:e:lhv";
static const char _SHORT_OPTS[] = "d:i:x:y:m:a:f:z:ntb:w:q:c:s:p:u:ro:k:e:lhv";
static const struct option _LONG_OPTS[] = {
{"device", required_argument, NULL, 'd'},
{"input", required_argument, NULL, 'i'},
@@ -82,6 +82,7 @@ static const struct option _LONG_OPTS[] = {
{"user", required_argument, NULL, 3000},
{"passwd", required_argument, NULL, 3001},
{"static", required_argument, NULL, 3002},
{"blank", required_argument, NULL, 'k'},
{"drop-same-frames", required_argument, NULL, 'e'},
{"slowdown", no_argument, NULL, 'l'},
{"fake-width", required_argument, NULL, 3003},
@@ -128,18 +129,19 @@ static void _help(struct device_t *dev, struct encoder_t *encoder, struct http_s
printf(" -z|--min-frame-size <N> ──────── Drop frames smaller then this limit. Useful if the device\n");
printf(" produces small-sized garbage frames. Default: disabled.\n\n");
printf(" -n|--persistent ──────────────── Don't re-initialize device on timeout. Default: disabled.\n\n");
printf(" -t|--dv-timings ──────────────── Enable DV timings queriyng and events processing.\n");
printf(" Supports automatic resolution changing. Default: disabled.\n\n");
printf(" -t|--dv-timings ──────────────── Enable DV timings queriyng and events processing\n");
printf(" to automatic resolution change. Default: disabled.\n\n");
printf(" -b|--buffers <N> ─────────────── The number of buffers to receive data from the device.\n");
printf(" Each buffer may processed using an intermediate thread.\n");
printf(" Default: %u (the number of CPU cores (but not more 4) + 1).\n\n", dev->n_buffers);
printf(" -w|--workers <N> ─────────────── The number of worker threads. Default: %u (== --buffers).\n\n", dev->n_workers);
printf(" Default: %u (the number of CPU cores (but not more than 4) + 1).\n\n", dev->n_buffers);
printf(" -w|--workers <N> ─────────────── The number of worker threads but not more than buffers.\n");
printf(" Default: %u (the number of CPU cores (but not more than 4)).\n\n", dev->n_workers);
printf(" -q|--quality <N> ─────────────── Set quality of JPEG encoding from 1 to 100 (best). Default: %u.\n\n", encoder->quality);
printf(" -c|--encoder <type> ──────────── Use specified encoder. It may affects to workers number.\n");
printf(" Available: %s; default: CPU.\n\n", ENCODER_TYPES_STR);
printf(" --device-timeout <seconds> ───── Timeout for device querying. Default: %u\n\n", dev->timeout);
printf(" --device-timeout <seconds> ───── Timeout for device querying. Default: %u.\n\n", dev->timeout);
printf(" --device-error-delay <seconds> ─ Delay before trying to connect to the device again\n");
printf(" after an error (timeout for example). Default: %u\n\n", dev->error_delay);
printf(" after an error (timeout for example). Default: %u.\n\n", dev->error_delay);
printf("Image control options:\n");
printf("══════════════════════\n");
printf(" --brightness <N> ───────────── Set brightness. Default: no change.\n\n");
@@ -166,6 +168,8 @@ static void _help(struct device_t *dev, struct encoder_t *encoder, struct http_s
printf(" --passwd <str> ───────────── HTTP basic auth passwd. Default: empty.\n\n");
printf(" --static <path> ───────────── Path to dir with static files instead of embedded root index page.\n");
printf(" Symlinks are not supported for security reasons. Default: disabled.\n\n");
printf(" -k|--blank <path> ─────────── Path to JPEG file that will be shown when the device is disconnected\n");
printf(" during the streaming. Default: black screen 640x480 with 'NO SIGNAL'.\n\n");
printf(" -e|--drop-same-frames <N> ── Don't send same frames to clients, but no more than specified number.\n");
printf(" It can significantly reduce the outgoing traffic, but will increase\n");
printf(" the CPU loading. Don't use this option with analog signal sources\n");
@@ -284,6 +288,7 @@ static int _parse_options(int argc, char *argv[], struct device_t *dev, struct e
case 3000: OPT_SET(server->user, optarg);
case 3001: OPT_SET(server->passwd, optarg);
case 3002: OPT_SET(server->static_path, optarg);
case 'k': OPT_SET(server->blank_path, optarg);
case 'e': OPT_UNSIGNED(server->drop_same_frames, "--drop-same-frames", 0, 30);
case 'l': OPT_SET(server->slowdown, true);
case 3003: OPT_UNSIGNED(server->fake_width, "--fake-width", 0, 1920);
@@ -379,7 +384,6 @@ int main(int argc, char *argv[]) {
if ((exit_code = _parse_options(argc, argv, dev, encoder, server)) == 0) {
_install_signal_handlers();
encoder_prepare(encoder, dev);
pthread_t stream_loop_tid;
pthread_t server_loop_tid;

View File

@@ -119,7 +119,7 @@ void stream_loop(struct stream_t *stream) {
LOG_PERF("##### Raw frame accepted; worker = %u", free_worker_number);
} else {
for (unsigned number = 0; number < stream->dev->n_workers; ++number) {
for (unsigned number = 0; number < stream->dev->run->n_workers; ++number) {
if (
!atomic_load(&pool.workers[number].has_job) && (
free_worker_number == -1
@@ -232,7 +232,7 @@ void stream_loop(struct stream_t *stream) {
LOG_DEBUG("Grabbed a new frame to buffer %u", buf_index);
if (!oldest_worker) {
if (oldest_worker == NULL) {
oldest_worker = &pool.workers[free_worker_number];
last_worker = oldest_worker;
} else {
@@ -315,7 +315,7 @@ static void _stream_expose_picture(struct stream_t *stream, unsigned buf_index)
stream->picture.used = PICTURE(used);
stream->picture.allocated = PICTURE(allocated);
memcpy(stream->picture.data, PICTURE(data), stream->picture.used * sizeof(*stream->picture.data));
memcpy(stream->picture.data, PICTURE(data), stream->picture.used);
stream->picture.grab_time = PICTURE(grab_time);
stream->picture.encode_begin_time = PICTURE(encode_begin_time);
@@ -336,7 +336,7 @@ static long double _stream_get_fluency_delay(struct device_t *dev, struct worker
long double min_delay;
long double soft_delay;
for (unsigned number = 0; number < dev->n_workers; ++number) {
for (unsigned number = 0; number < dev->run->n_workers; ++number) {
# define WORKER(_next) pool->workers[number]._next
A_MUTEX_LOCK(&WORKER(last_comp_time_mutex));
@@ -347,9 +347,9 @@ static long double _stream_get_fluency_delay(struct device_t *dev, struct worker
# undef WORKER
}
avg_comp_time = sum_comp_time / dev->n_workers; // Среднее время работы воркеров
avg_comp_time = sum_comp_time / dev->run->n_workers; // Среднее время работы воркеров
min_delay = avg_comp_time / dev->n_workers; // Среднее время работы размазывается на N воркеров
min_delay = avg_comp_time / dev->run->n_workers; // Среднее время работы размазывается на N воркеров
if (dev->desired_fps > 0 && min_delay > 0) {
// Искусственное время задержки на основе желаемого FPS, если включен --desired-fps
@@ -389,7 +389,7 @@ static int _stream_init(struct stream_t *stream, struct workers_pool_t *pool) {
goto error;
}
encoder_prepare_live(pool->encoder, stream->dev);
encoder_prepare(pool->encoder, stream->dev);
_stream_init_workers(stream, pool);
@@ -401,15 +401,15 @@ static int _stream_init(struct stream_t *stream, struct workers_pool_t *pool) {
}
static void _stream_init_workers(struct stream_t *stream, struct workers_pool_t *pool) {
LOG_INFO("Spawning %u workers ...", stream->dev->n_workers);
LOG_INFO("Spawning %u workers ...", stream->dev->run->n_workers);
atomic_store(&pool->workers_stop, false);
A_CALLOC(pool->workers, stream->dev->n_workers);
A_CALLOC(pool->workers, stream->dev->run->n_workers);
A_MUTEX_INIT(&pool->free_workers_mutex);
A_COND_INIT(&pool->free_workers_cond);
for (unsigned number = 0; number < stream->dev->n_workers; ++number) {
for (unsigned number = 0; number < stream->dev->run->n_workers; ++number) {
# define WORKER(_next) pool->workers[number]._next
pool->free_workers += 1;
@@ -492,7 +492,7 @@ static void _stream_destroy_workers(struct stream_t *stream, struct workers_pool
LOG_INFO("Destroying workers ...");
atomic_store(&pool->workers_stop, true);
for (unsigned number = 0; number < stream->dev->n_workers; ++number) {
for (unsigned number = 0; number < stream->dev->run->n_workers; ++number) {
# define WORKER(_next) pool->workers[number]._next
A_MUTEX_LOCK(&WORKER(has_job_mutex));