/*
* Copyright (C) 2018 Red Hat Inc.
*
* 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 2 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, write to the Free Software
* Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
* 02111-1307, USA.
*
*/
#include "config.h"
#include "grd-vnc-pipewire-stream.h"
#include <linux/dma-buf.h>
#include <pipewire/pipewire.h>
#include <spa/param/props.h>
#include <spa/param/format-utils.h>
#include <spa/param/video/format-utils.h>
#include <spa/utils/result.h>
#include <sys/ioctl.h>
#include <sys/mman.h>
#include <sys/syscall.h>
#include "grd-vnc-cursor.h"
enum
{
CLOSED,
N_SIGNALS
};
guint signals[N_SIGNALS];
typedef struct _GrdPipeWireSource
{
GSource base;
struct pw_loop *pipewire_loop;
} GrdPipeWireSource;
typedef struct _GrdVncFrame
{
void *data;
rfbCursorPtr rfb_cursor;
gboolean cursor_moved;
int cursor_x;
int cursor_y;
} GrdVncFrame;
struct _GrdVncPipeWireStream
{
GObject parent;
GrdSessionVnc *session;
GSource *pipewire_source;
struct pw_context *pipewire_context;
struct pw_core *pipewire_core;
struct spa_hook pipewire_core_listener;
GMutex frame_mutex;
GrdVncFrame *pending_frame;
struct pw_stream *pipewire_stream;
struct spa_hook pipewire_stream_listener;
uint32_t src_node_id;
struct spa_video_info_raw spa_format;
};
G_DEFINE_TYPE (GrdVncPipeWireStream, grd_vnc_pipewire_stream,
G_TYPE_OBJECT)
#define CURSOR_META_SIZE(width, height) \
(sizeof(struct spa_meta_cursor) + \
sizeof(struct spa_meta_bitmap) + width * height * 4)
static gboolean
pipewire_loop_source_prepare (GSource *base,
int *timeout)
{
*timeout = -1;
return FALSE;
}
static gboolean
pipewire_loop_source_dispatch (GSource *source,
GSourceFunc callback,
gpointer user_data)
{
GrdPipeWireSource *pipewire_source = (GrdPipeWireSource *) source;
int result;
result = pw_loop_iterate (pipewire_source->pipewire_loop, 0);
if (result < 0)
g_warning ("pipewire_loop_iterate failed: %s", spa_strerror (result));
return TRUE;
}
static void
pipewire_loop_source_finalize (GSource *source)
{
GrdPipeWireSource *pipewire_source = (GrdPipeWireSource *) source;
pw_loop_leave (pipewire_source->pipewire_loop);
pw_loop_destroy (pipewire_source->pipewire_loop);
}
static GSourceFuncs pipewire_source_funcs =
{
pipewire_loop_source_prepare,
NULL,
pipewire_loop_source_dispatch,
pipewire_loop_source_finalize
};
static GrdPipeWireSource *
create_pipewire_source (void)
{
GrdPipeWireSource *pipewire_source;
pipewire_source =
(GrdPipeWireSource *) g_source_new (&pipewire_source_funcs,
sizeof (GrdPipeWireSource));
pipewire_source->pipewire_loop = pw_loop_new (NULL);
if (!pipewire_source->pipewire_loop)
{
g_source_destroy ((GSource *) pipewire_source);
return NULL;
}
g_source_add_unix_fd (&pipewire_source->base,
pw_loop_get_fd (pipewire_source->pipewire_loop),
G_IO_IN | G_IO_ERR);
pw_loop_enter (pipewire_source->pipewire_loop);
g_source_attach (&pipewire_source->base, NULL);
return pipewire_source;
}
static void
on_stream_state_changed (void *user_data,
enum pw_stream_state old,
enum pw_stream_state state,
const char *error)
{
g_debug ("Pipewire stream state changed from %s to %s",
pw_stream_state_as_string (old),
pw_stream_state_as_string (state));
switch (state)
{
case PW_STREAM_STATE_ERROR:
g_warning ("PipeWire stream error: %s", error);
break;
case PW_STREAM_STATE_PAUSED:
case PW_STREAM_STATE_STREAMING:
case PW_STREAM_STATE_UNCONNECTED:
case PW_STREAM_STATE_CONNECTING:
break;
}
}
static void
on_stream_param_changed (void *user_data,
uint32_t id,
const struct spa_pod *format)
{
GrdVncPipeWireStream *stream = GRD_VNC_PIPEWIRE_STREAM (user_data);
uint8_t params_buffer[1024];
struct spa_pod_builder pod_builder;
int width;
int height;
const struct spa_pod *params[3];
if (!format || id != SPA_PARAM_Format)
return;
spa_format_video_raw_parse (format, &stream->spa_format);
pod_builder = SPA_POD_BUILDER_INIT (params_buffer, sizeof (params_buffer));
width = stream->spa_format.size.width;
height = stream->spa_format.size.height;
grd_session_vnc_queue_resize_framebuffer (stream->session, width, height);
params[0] = spa_pod_builder_add_object (
&pod_builder,
SPA_TYPE_OBJECT_ParamBuffers, SPA_PARAM_Buffers,
SPA_PARAM_BUFFERS_buffers, SPA_POD_CHOICE_RANGE_Int (8, 1, 8),
0);
params[1] = spa_pod_builder_add_object (
&pod_builder,
SPA_TYPE_OBJECT_ParamMeta, SPA_PARAM_Meta,
SPA_PARAM_META_type, SPA_POD_Id (SPA_META_Header),
SPA_PARAM_META_size, SPA_POD_Int (sizeof (struct spa_meta_header)),
0);
params[2] = spa_pod_builder_add_object(
&pod_builder,
SPA_TYPE_OBJECT_ParamMeta, SPA_PARAM_Meta,
SPA_PARAM_META_type, SPA_POD_Id (SPA_META_Cursor),
SPA_PARAM_META_size, SPA_POD_CHOICE_RANGE_Int (CURSOR_META_SIZE (64,64),
CURSOR_META_SIZE (1,1),
CURSOR_META_SIZE (256,256)),
0);
pw_stream_update_params (stream->pipewire_stream,
params, G_N_ELEMENTS (params));
}
static gboolean
spa_pixel_format_to_grd_pixel_format (uint32_t spa_format,
GrdPixelFormat *out_format)
{
if (spa_format == SPA_VIDEO_FORMAT_RGBA)
*out_format = GRD_PIXEL_FORMAT_RGBA8888;
else
return FALSE;
return TRUE;
}
static void
sync_dma_buf (int fd,
uint64_t start_or_end)
{
struct dma_buf_sync sync = { 0 };
sync.flags = start_or_end | DMA_BUF_SYNC_READ;
while (TRUE)
{
int ret;
ret = ioctl (fd, DMA_BUF_IOCTL_SYNC, &sync);
if (ret == -1 && errno == EINTR)
{
continue;
}
else if (ret == -1)
{
g_warning ("Failed to synchronize DMA buffer: %s",
g_strerror (errno));
break;
}
else
{
break;
}
}
}
static int
do_render (struct spa_loop *loop,
bool async,
uint32_t seq,
const void *data,
size_t size,
void *user_data)
{
GrdVncPipeWireStream *stream = GRD_VNC_PIPEWIRE_STREAM (user_data);
GrdVncFrame *frame;
g_mutex_lock (&stream->frame_mutex);
frame = g_steal_pointer (&stream->pending_frame);
g_mutex_unlock (&stream->frame_mutex);
if (!frame)
return 0;
if (frame->rfb_cursor)
grd_session_vnc_set_cursor (stream->session, frame->rfb_cursor);
if (frame->cursor_moved)
{
grd_session_vnc_move_cursor (stream->session,
frame->cursor_x,
frame->cursor_y);
}
if (frame->data)
grd_session_vnc_take_buffer (stream->session, frame->data);
else
grd_session_vnc_flush (stream->session);
g_free (frame);
return 0;
}
static GrdVncFrame *
process_buffer (GrdVncPipeWireStream *stream,
struct spa_buffer *buffer)
{
size_t size;
uint8_t *map;
void *src_data;
struct spa_meta_cursor *spa_meta_cursor;
g_autofree GrdVncFrame *frame = NULL;
frame = g_new0 (GrdVncFrame, 1);
if (buffer->datas[0].chunk->size == 0)
{
map = NULL;
src_data = NULL;
}
else if (buffer->datas[0].type == SPA_DATA_MemFd)
{
size = buffer->datas[0].maxsize + buffer->datas[0].mapoffset;
map = mmap (NULL, size, PROT_READ, MAP_PRIVATE, buffer->datas[0].fd, 0);
if (map == MAP_FAILED)
{
g_warning ("Failed to mmap buffer: %s", g_strerror (errno));
return NULL;
}
src_data = SPA_MEMBER (map, buffer->datas[0].mapoffset, uint8_t);
}
else if (buffer->datas[0].type == SPA_DATA_DmaBuf)
{
int fd;
fd = buffer->datas[0].fd;
size = buffer->datas[0].maxsize + buffer->datas[0].mapoffset;
map = mmap (NULL, size, PROT_READ, MAP_PRIVATE, fd, 0);
if (map == MAP_FAILED)
{
g_warning ("Failed to mmap DMA buffer: %s", g_strerror (errno));
return NULL;
}
sync_dma_buf (fd, DMA_BUF_SYNC_START);
src_data = SPA_MEMBER (map, buffer->datas[0].mapoffset, uint8_t);
}
else if (buffer->datas[0].type == SPA_DATA_MemPtr)
{
size = buffer->datas[0].maxsize + buffer->datas[0].mapoffset;
map = NULL;
src_data = buffer->datas[0].data;
}
else
{
return NULL;
}
if (src_data)
{
int src_stride;
int dst_stride;
int height;
int y;
src_stride = buffer->datas[0].chunk->stride;
dst_stride = grd_session_vnc_get_framebuffer_stride (stream->session);
height = stream->spa_format.size.height;
frame->data = g_malloc (height * dst_stride);
for (y = 0; y < height; y++)
{
memcpy (((uint8_t *) frame->data) + y * dst_stride,
((uint8_t *) src_data) + y * src_stride,
dst_stride);
}
}
if (map)
{
if (buffer->datas[0].type == SPA_DATA_DmaBuf)
sync_dma_buf (buffer->datas[0].fd, DMA_BUF_SYNC_END);
munmap (map, size);
}
spa_meta_cursor = spa_buffer_find_meta_data (buffer, SPA_META_Cursor,
sizeof *spa_meta_cursor);
if (spa_meta_cursor && spa_meta_cursor_is_valid (spa_meta_cursor))
{
struct spa_meta_bitmap *spa_meta_bitmap;
GrdPixelFormat format;
if (spa_meta_cursor->bitmap_offset)
{
spa_meta_bitmap = SPA_MEMBER (spa_meta_cursor,
spa_meta_cursor->bitmap_offset,
struct spa_meta_bitmap);
}
else
{
spa_meta_bitmap = NULL;
}
if (spa_meta_bitmap &&
spa_meta_bitmap->size.width > 0 &&
spa_meta_bitmap->size.height > 0 &&
spa_pixel_format_to_grd_pixel_format (spa_meta_bitmap->format,
&format))
{
uint8_t *buf;
rfbCursorPtr rfb_cursor;
buf = SPA_MEMBER (spa_meta_bitmap, spa_meta_bitmap->offset, uint8_t);
rfb_cursor = grd_vnc_create_cursor (spa_meta_bitmap->size.width,
spa_meta_bitmap->size.height,
spa_meta_bitmap->stride,
format,
buf);
rfb_cursor->xhot = spa_meta_cursor->hotspot.x;
rfb_cursor->yhot = spa_meta_cursor->hotspot.y;
frame->rfb_cursor = rfb_cursor;
}
else if (spa_meta_bitmap)
{
frame->rfb_cursor = grd_vnc_create_empty_cursor (1, 1);
}
frame->cursor_moved = TRUE;
frame->cursor_x = spa_meta_cursor->position.x;
frame->cursor_y = spa_meta_cursor->position.y;
}
return g_steal_pointer (&frame);
}
static void
on_stream_process (void *user_data)
{
GrdVncPipeWireStream *stream = GRD_VNC_PIPEWIRE_STREAM (user_data);
GrdPipeWireSource *pipewire_source =
(GrdPipeWireSource *) stream->pipewire_source;
struct pw_buffer *next_buffer;
struct pw_buffer *buffer = NULL;
GrdVncFrame *frame;
next_buffer = pw_stream_dequeue_buffer (stream->pipewire_stream);
while (next_buffer)
{
buffer = next_buffer;
next_buffer = pw_stream_dequeue_buffer (stream->pipewire_stream);
if (next_buffer)
pw_stream_queue_buffer (stream->pipewire_stream, buffer);
}
if (!buffer)
return;
frame = process_buffer (stream, buffer->buffer);
g_mutex_lock (&stream->frame_mutex);
if (stream->pending_frame)
{
g_free (stream->pending_frame->data);
g_clear_pointer (&stream->pending_frame, g_free);
}
stream->pending_frame = frame;
g_mutex_unlock (&stream->frame_mutex);
pw_stream_queue_buffer (stream->pipewire_stream, buffer);
pw_loop_invoke (pipewire_source->pipewire_loop, do_render,
SPA_ID_INVALID, NULL, 0,
false, stream);
}
static const struct pw_stream_events stream_events = {
PW_VERSION_STREAM_EVENTS,
.state_changed = on_stream_state_changed,
.param_changed = on_stream_param_changed,
.process = on_stream_process,
};
static gboolean
connect_to_stream (GrdVncPipeWireStream *stream,
GError **error)
{
struct pw_stream *pipewire_stream;
uint8_t params_buffer[1024];
struct spa_pod_builder pod_builder;
struct spa_rectangle min_rect;
struct spa_rectangle max_rect;
struct spa_fraction min_framerate;
struct spa_fraction max_framerate;
const struct spa_pod *params[2];
int ret;
pipewire_stream = pw_stream_new (stream->pipewire_core,
"grd-vnc-pipewire-stream",
NULL);
min_rect = SPA_RECTANGLE (1, 1);
max_rect = SPA_RECTANGLE (INT32_MAX, INT32_MAX);
min_framerate = SPA_FRACTION (1, 1);
max_framerate = SPA_FRACTION (30, 1);
pod_builder = SPA_POD_BUILDER_INIT (params_buffer, sizeof (params_buffer));
params[0] = spa_pod_builder_add_object (
&pod_builder,
SPA_TYPE_OBJECT_Format, SPA_PARAM_EnumFormat,
SPA_FORMAT_mediaType, SPA_POD_Id (SPA_MEDIA_TYPE_video),
SPA_FORMAT_mediaSubtype, SPA_POD_Id (SPA_MEDIA_SUBTYPE_raw),
SPA_FORMAT_VIDEO_format, SPA_POD_Id (SPA_VIDEO_FORMAT_BGRx),
SPA_FORMAT_VIDEO_size, SPA_POD_CHOICE_RANGE_Rectangle (&min_rect,
&min_rect,
&max_rect),
SPA_FORMAT_VIDEO_framerate, SPA_POD_Fraction (&SPA_FRACTION(0, 1)),
SPA_FORMAT_VIDEO_maxFramerate, SPA_POD_CHOICE_RANGE_Fraction (&min_framerate,
&min_framerate,
&max_framerate),
0);
stream->pipewire_stream = pipewire_stream;
pw_stream_add_listener (pipewire_stream,
&stream->pipewire_stream_listener,
&stream_events,
stream);
ret = pw_stream_connect (stream->pipewire_stream,
PW_DIRECTION_INPUT,
stream->src_node_id,
(PW_STREAM_FLAG_RT_PROCESS |
PW_STREAM_FLAG_AUTOCONNECT),
params, 1);
if (ret < 0)
{
g_set_error_literal (error, G_IO_ERROR, g_io_error_from_errno (-ret),
strerror (-ret));
return FALSE;
}
return TRUE;
}
static void
on_core_error (void *user_data,
uint32_t id,
int seq,
int res,
const char *message)
{
GrdVncPipeWireStream *stream = GRD_VNC_PIPEWIRE_STREAM (user_data);
g_warning ("Pipewire core error: id:%u %s", id, message);
if (id == PW_ID_CORE && res == -EPIPE)
g_signal_emit (stream, signals[CLOSED], 0);
}
static const struct pw_core_events core_events = {
PW_VERSION_CORE_EVENTS,
.error = on_core_error,
};
GrdVncPipeWireStream *
grd_vnc_pipewire_stream_new (GrdSessionVnc *session_vnc,
uint32_t src_node_id,
GError **error)
{
g_autoptr (GrdVncPipeWireStream) stream = NULL;
GrdPipeWireSource *pipewire_source;
static gboolean is_pipewire_initialized = FALSE;
if (!is_pipewire_initialized)
{
pw_init (NULL, NULL);
is_pipewire_initialized = TRUE;
}
stream = g_object_new (GRD_TYPE_VNC_PIPEWIRE_STREAM, NULL);
stream->session = session_vnc;
stream->src_node_id = src_node_id;
pipewire_source = create_pipewire_source ();
if (!pipewire_source)
{
g_set_error (error, G_IO_ERROR, G_IO_ERROR_FAILED,
"Failed to create PipeWire source");
return NULL;
}
stream->pipewire_source = (GSource *) pipewire_source;
stream->pipewire_context = pw_context_new (pipewire_source->pipewire_loop,
NULL, 0);
if (!stream->pipewire_context)
{
g_set_error (error, G_IO_ERROR, G_IO_ERROR_FAILED,
"Failed to create pipewire context");
return NULL;
}
stream->pipewire_core = pw_context_connect (stream->pipewire_context, NULL, 0);
if (!stream->pipewire_core)
{
g_set_error (error, G_IO_ERROR, G_IO_ERROR_FAILED,
"Failed to connect pipewire context");
return NULL;
}
pw_core_add_listener (stream->pipewire_core,
&stream->pipewire_core_listener,
&core_events,
stream);
if (!connect_to_stream (stream, error))
return NULL;
return g_steal_pointer (&stream);
}
static void
grd_vnc_pipewire_stream_finalize (GObject *object)
{
GrdVncPipeWireStream *stream = GRD_VNC_PIPEWIRE_STREAM (object);
/*
* We can't clear stream->pipewire_stream before destroying it, as the data
* thread in PipeWire might access the variable during destruction.
*/
if (stream->pipewire_stream)
pw_stream_destroy (stream->pipewire_stream);
g_clear_pointer (&stream->pipewire_core, pw_core_disconnect);
g_clear_pointer (&stream->pipewire_context, pw_context_destroy);
g_clear_pointer (&stream->pipewire_source, g_source_destroy);
G_OBJECT_CLASS (grd_vnc_pipewire_stream_parent_class)->finalize (object);
}
static void
grd_vnc_pipewire_stream_init (GrdVncPipeWireStream *stream)
{
g_mutex_init (&stream->frame_mutex);
}
static void
grd_vnc_pipewire_stream_class_init (GrdVncPipeWireStreamClass *klass)
{
GObjectClass *object_class = G_OBJECT_CLASS (klass);
object_class->finalize = grd_vnc_pipewire_stream_finalize;
signals[CLOSED] = g_signal_new ("closed",
G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST,
0,
NULL, NULL, NULL,
G_TYPE_NONE, 0);
}