core: Set timeout for requests

Changes

6 changed files (+64/-28)

• core/src/browse.zig
  • @@ -486,6 +486,7 @@ missing_property = 2,
    out_of_memory = 3, failed_to_send = 4, closed = 5, + timeout = 6, }; pub const ResultAction = enum(c_int) {
• core/src/connection.zig
  • @@ -332,14 +332,19 @@ Event.exportEventCastFunction(transport.ZoneListEvent, .zone_list);
    } }; - fn Listener(comptime T: type) type { + inline fn timedWaitMs(reset_event: *std.Thread.ResetEvent, timeout_ms: usize) error{Timeout}!void { + // 1 milliseconds = 1,000 microseconds + // 1 microseconds = 1,000 nanoseconds + try reset_event.timedWait(timeout_ms * 1000 * 1000); + } + + fn Listener(comptime T: type, timeout_ms: usize) type { return struct { wrote: std.Thread.ResetEvent = .{}, data: ?T = null, - pub fn listen(self: *@This()) T { - self.wrote.wait(); - + pub fn listen(self: *@This()) error{Timeout}!T { + try timedWaitMs(&self.wrote, timeout_ms); return self.data orelse @panic("Set Listener.wrote before writing data"); }
  • @@ -350,8 +355,8 @@ }
    }; } - fn JsonResponseListener(comptime T: type) type { - return Listener(moo.JsonBody(T)); + fn JsonResponseListener(comptime T: type, timeout_ms: usize) type { + return Listener(moo.JsonBody(T), timeout_ms); } const ImageDownload = struct {
  • @@ -376,6 +381,11 @@ ) @This() {
    return .{ .hash = ImageDownload.getHash(image_key, opts) }; } + pub fn getDownloaded(self: *@This()) error{Timeout}!*image.GetResult { + try timedWaitMs(&self.ready, 5_000); + return self.downloaded; + } + pub fn write(self: *@This(), data: *image.GetResult) void { self.downloaded = data.retain(); self.ready.set();
  • @@ -392,9 +402,9 @@ pub const Connection = extern struct {
    const cname = "plac_connection"; const allocator = std.heap.c_allocator; - const BrowseListener = JsonResponseListener(BrowseService.Browse.Response); - const LoadListener = JsonResponseListener(BrowseService.Load.Response); - const ControlListener = Listener(transport.ControlResultCode); + const BrowseListener = JsonResponseListener(BrowseService.Browse.Response, 1_000); + const LoadListener = JsonResponseListener(BrowseService.Load.Response, 3_000); + const ControlListener = Listener(transport.ControlResultCode, 2_000); internal: *Internal,
  • @@ -410,7 +420,7 @@ arc: Arc = .{},
    saved_token: ?[]const u8 = null, browse_listeners: std.AutoHashMap(u64, BrowseListener), load_listeners: std.AutoHashMap(u64, LoadListener), - image_downloads: std.AutoHashMap(u64, ImageDownload), + image_downloads: std.AutoHashMap(u64, *ImageDownload), image_downloads_lock: std.Thread.Mutex = .{}, control_events: std.AutoHashMap(u64, ControlListener),
  • @@ -442,7 +452,7 @@ .host = host,
    .saved_token = saved_token, .browse_listeners = std.AutoHashMap(u64, BrowseListener).init(allocator), .load_listeners = std.AutoHashMap(u64, LoadListener).init(allocator), - .image_downloads = std.AutoHashMap(u64, ImageDownload).init(allocator), + .image_downloads = std.AutoHashMap(u64, *ImageDownload).init(allocator), .control_events = std.AutoHashMap(u64, ControlListener).init(allocator), }; }
  • @@ -453,7 +463,7 @@ self.load_listeners.deinit();
    var image_downloads = self.image_downloads.valueIterator(); while (image_downloads.next()) |download| { - download.deinit(); + download.*.deinit(); } self.image_downloads.deinit(); self.control_events.deinit();
  • @@ -706,7 +716,7 @@ msg.data,
    ) catch |err| { std.log.debug("Received unexpected get_image response: {s}", .{@errorName(err)}); - entry.value_ptr.write( + entry.value_ptr.*.write( image.GetResult.makeError(.unexpected_response) catch |obj_err| { std.log.err("Unable to compose image.GetResult: {s}", .{@errorName(obj_err)}); continue;
  • @@ -715,7 +725,7 @@ );
    continue; }; - entry.value_ptr.write(image.GetResult.make(&res) catch |err| { + entry.value_ptr.*.write(image.GetResult.make(&res) catch |err| { std.log.err("Unable to compose image.GetResult: {s}", .{@errorName(err)}); continue; });
  • @@ -991,7 +1001,9 @@ std.log.err("Unable to write control request: {s}", .{@errorName(err)});
    return .failed_to_send; }; - return entry.value_ptr.listen(); + return entry.value_ptr.listen() catch { + return .timeout; + }; } pub fn requestBrowse(
  • @@ -1047,7 +1059,10 @@ };
    std.log.debug("Sent browse request (ID={d})", .{req_id}); - const resp = entry.value_ptr.listen(); + const resp = entry.value_ptr.listen() catch { + std.log.err("Browse request timeout", .{}); + return browse.Result.makeRetainedError(.timeout) catch null; + }; defer resp.deinit(); switch (resp.value.action) {
  • @@ -1118,7 +1133,10 @@ };
    std.log.debug("Sent load request (ID={d})", .{load_req_id}); - const load_resp = load_entry.value_ptr.listen(); + const load_resp = load_entry.value_ptr.listen() catch { + std.log.err("Load request timeout", .{}); + return browse.Result.makeRetainedError(.timeout) catch null; + }; defer load_resp.deinit(); const action = browse.ListAction.make(load_resp.value) catch {
  • @@ -1154,10 +1172,14 @@ defer self.internal.image_downloads_lock.unlock();
    var download_iter = self.internal.image_downloads.valueIterator(); while (download_iter.next()) |download| { - if (download.hash == req_hash) { + if (download.*.hash == req_hash) { std.log.debug("Using existing request for image_key={s}", .{image_key}); - download.ready.wait(); - return download.downloaded.retain(); + const downloaded = download.*.getDownloaded() catch { + std.log.err("get_image timeout", .{}); + return image.GetResult.makeRetainedError(.timeout) catch null; + }; + + return downloaded.retain(); } } }
  • @@ -1196,21 +1218,22 @@ return image.GetResult.makeRetainedError(.unknown_error) catch null;
    }; defer allocator.free(req_msg); - var entry = entry: { + var download = ImageDownload.init(image_key, opts); + + { self.internal.image_downloads_lock.lock(); defer self.internal.image_downloads_lock.unlock(); - const ent = self.internal.image_downloads.getOrPut(req_id) catch |err| { + self.internal.image_downloads.put(req_id, &download) catch |err| { std.log.err("Unable to set listener for get_image response: {s}", .{@errorName(err)}); return image.GetResult.makeRetainedError(.unknown_error) catch null; }; - ent.value_ptr.* = ImageDownload.init(image_key, opts); - break :entry ent; - }; + } + defer { self.internal.image_downloads_lock.lock(); defer self.internal.image_downloads_lock.unlock(); - entry.value_ptr.deinit(); + download.deinit(); _ = self.internal.image_downloads.remove(req_id); }
  • @@ -1221,8 +1244,12 @@ };
    std.log.debug("Sent get_image request (ID={d})", .{req_id}); - entry.value_ptr.ready.wait(); - return entry.value_ptr.downloaded.retain(); + const downloaded = download.getDownloaded() catch { + std.log.err("get_image timeout", .{}); + return image.GetResult.makeRetainedError(.timeout) catch null; + }; + + return downloaded.retain(); } pub fn export_capi() void {
• core/src/image.zig
  • @@ -38,6 +38,7 @@ out_of_memory = 2,
    unexpected_response = 3, socket_closed = 4, failed_to_send = 5, + timeout = 6, }; pub const GetOptions = extern struct {
• core/src/plac.h
  • @@ -143,6 +143,7 @@ PLAC_TRANSPORT_CONTROL_RESULT_FAILED_TO_SEND = 3,
    PLAC_TRANSPORT_CONTROL_RESULT_CLOSED = 4, PLAC_TRANSPORT_CONTROL_RESULT_NO_ACTION_BIT_SET = 5, PLAC_TRANSPORT_CONTROL_RESULT_SERVER_ERROR = 6, + PLAC_TRANSPORT_CONTROL_RESULT_TIMEOUT = 7, } plac_transport_control_result_code; // browse.Hierarchy
  • @@ -199,6 +200,7 @@ PLAC_BROWSE_RESULT_MISSING_PROPERTY = 2,
    PLAC_BROWSE_RESULT_OUT_OF_MEMORY = 3, PLAC_BROWSE_RESULT_FAILED_TO_SEND = 4, PLAC_BROWSE_RESULT_CLOSED = 5, + PLAC_BROWSE_RESULT_TIMEOUT = 6, } plac_browse_result_code; // browse.ResultAction
  • @@ -282,6 +284,7 @@ PLAC_IMAGE_GET_RESULT_OUT_OF_MEMORY = 2,
    PLAC_IMAGE_GET_RESULT_UNEXPECTED_RESPONSE = 3, PLAC_IMAGE_GET_RESULT_SOCKET_CLOSED = 4, PLAC_IMAGE_GET_RESULT_FAILED_TO_SEND = 5, + PLAC_IMAGE_GET_RESULT_TIMEOUT = 6, } plac_image_get_result_code; // image.GetOptions
• core/src/plac.vapi
  • @@ -214,6 +214,7 @@ FAILED_TO_SEND = 3,
    CLOSED = 4, NO_ACTION_BIT_SET = 5, SERVER_ERROR = 6, + TIMEOUT = 7, } }
  • @@ -288,6 +289,7 @@ MISSING_PROPERTY = 2,
    OUT_OF_MEMORY = 3, FAILED_TO_SEND = 4, CLOSED = 5, + TIMEOUT = 6, } [CCode (
  • @@ -418,6 +420,7 @@ OUT_OF_MEMORY = 2,
    UNEXPECTED_RESPONSE = 3, SOCKET_CLOSED = 4, FAILED_TO_SEND = 5, + TIMEOUT = 6, } [CCode (
• core/src/transport.zig
  • @@ -507,6 +507,7 @@ failed_to_send = 3,
    closed = 4, no_action_bit_set = 5, server_error = 6, + timeout = 7, }; pub fn export_capi() void {