# # xsequencer -ximp.pkg
#
# For the big picture see the imp dataflow diagrams in
#
# src/lib/x-kit/xclient/src/window/xclient-ximps.pkg#
# The sequencer is responsible for matching
# replies read from the X with requests sent
# to it.
#
# All requests to the X-server go through the sequencer,
# as do all replies from the X-server.
#
# The sequencer communicates on five fixed channels:
#
# plea_mailslot -- request messages from clients
# from_x_mailslot -- reply, error and event messages from the X server (via the input buffer)
# to_x_mailslot -- requests messages to the X server (via the output buffer)
# xevent_mailslot -- X-events to the X-event buffer (and thence to clients)
# error_sink_mailslot -- errors to the error handler
#
# In addition, the sequencer sends replies
# to clients on the reply channel that was
# bundled with the request.
#
# We maintain a pending-reply queue of requests sent
# to the X server for which replies are expected but
# not yet received.
# We represent it using the usual two-list arrangement,
# writing new entries to the rear list while reading them
# from the front list; when the front list is empty we
# reverse the rear list and make it the new front list.
# Compiled by:
# src/lib/x-kit/xclient/xclient-internals.sublibstipulate
include package threadkit; # threadkit is from src/lib/src/lib/thread-kit/src/core-thread-kit/threadkit.pkg #
#
package un = unt; # unt is from src/lib/std/unt.pkg package v1u = vector_of_one_byte_unts; # vector_of_one_byte_unts is from src/lib/std/src/vector-of-one-byte-unts.pkg package w2v = wire_to_value; # wire_to_value is from src/lib/x-kit/xclient/src/wire/wire-to-value.pkg package v2w = value_to_wire; # value_to_wire is from src/lib/x-kit/xclient/src/wire/value-to-wire.pkg package e2s = xerror_to_string; # xerror_to_string is from src/lib/x-kit/xclient/src/to-string/xerror-to-string.pkg package g2d = geometry2d; # geometry2d is from src/lib/std/2d/geometry2d.pkg package xtr = xlogger; # xlogger is from src/lib/x-kit/xclient/src/stuff/xlogger.pkg package xew = xerror_well; # xerror_well is from src/lib/x-kit/xclient/src/wire/xerror-well.pkg package op = xsequencer_to_outbuf; # xsequencer_to_outbuf is from src/lib/x-kit/xclient/src/wire/xsequencer-to-outbuf.pkg package x2s = xclient_to_sequencer; # xclient_to_sequencer is from src/lib/x-kit/xclient/src/wire/xclient-to-sequencer.pkg package xps = xpacket_sink; # xpacket_sink is from src/lib/x-kit/xclient/src/wire/xpacket-sink.pkg package xt = xtypes; # xtypes is from src/lib/x-kit/xclient/src/wire/xtypes.pkg package xet = xevent_types; # xevent_types is from src/lib/x-kit/xclient/src/wire/xevent-types.pkg #
trace = xtr::log_if xtr::io_logging 0; # Conditionally write strings to tracing.log or whatever.
herein
# This imp is typically instantiated by:
#
# src/lib/x-kit/xclient/src/wire/xsocket-ximps.pkg package xsequencer_ximp
: (weak) Xsequencer_Ximp # Xsequencer_Ximp is from src/lib/x-kit/xclient/src/wire/xsequencer-ximp.api {
exception LOST_REPLY;
exception ERROR_REPLY xerrors::Xerror;
max_bytes_per_socket_write = 2048;
Xpacket_Plea = XPLEA_NOTE_XPACKET xps::Xpacket;
# Sequencer replies to client requests:
#
# Client pleas to sequencer:
#
Client_Plea #
= PLEA_SEND_BYTEVECTOR v1u::Vector
| PLEA_SEND_BYTEVECTORS List( v1u::Vector )
| PLEA_REPLY (v1u::Vector, Oneshot_Maildrop(x2s::Reply_Mail))
| PLEA_AND_CHECK (v1u::Vector, Oneshot_Maildrop(x2s::Reply_Mail))
# = PLEA_QUIT
# | PLEA_REPLIES (v1u::Vector, Mailslot(x2s::Reply_Mail), (v1u::Vector -> Int))
;
Pending_Reply = ONE_REPLY (un::Unt, Oneshot_Maildrop( x2s::Reply_Mail ))
| EXPOSURE_REPLY (un::Unt, Oneshot_Maildrop( Void -> List( g2d::Box ) ))
| ERROR_CHECK (un::Unt, Oneshot_Maildrop( x2s::Reply_Mail ))
# | MULTI_REPLY (un::Unt, Oneshot_Maildrop( x2s::Reply_Mail ), (v1u::Vector -> Int), List( v1u::Vector ))
;
#
# Above gives the kind of reply that is
# pending for an outstanding request in
# the outstanding-request queue.
#
# We use unsigneds to represent the
# sequence numbers.
#
# ONE_REPLY is the workhorse call:
# A request generating a single reply.
#
# MULTI_REPLY is a currently unused call
# supporting multiple responses to a single request:
# we accumulate responses until the (v1u::Vector -> Int)
# function argument ("remaining") returns 0.
# (The fourth slot is just the reply accumulator.)
Xsequencer_Ximp_State # Holds all nonephemeral mutable state maintained by ximp.
=
{ last_seqn_read: Ref(Unt),
last_seqn_sent: Ref(Unt),
#
pending_reply_queue: Ref { front: List( Pending_Reply ),
rear: List( Pending_Reply )
}
};
Imports = { # PUBLIC.
xsequencer_to_outbuf: op::Xsequencer_To_Outbuf, #
xpacket_sink: xps::Xpacket_Sink # Carries xpackets to decode_xpackets_ximp from src/lib/x-kit/xclient/src/wire/decode-xpackets-ximp.pkg };
Me_Slot = Mailslot( { imports: Imports,
me: Xsequencer_Ximp_State,
run_gun': Run_Gun,
end_gun': End_Gun
}
);
Xpacket_Q = Mailqueue( Xpacket_Plea );
Xerror_Q = Mailqueue( x2s::Xerror );
Exports = { # PUBLIC.
xpacket_sink: xps::Xpacket_Sink, # Carries xpackets to us from xserver via inbuf-ximp.
xclient_to_sequencer: x2s::Xclient_To_Sequencer, # Requests from widget/application code.
xerror_well: xew::Xerror_Well # Error messages from the X server.
};
Option = MICROTHREAD_NAME String; # PUBLIC.
Xsequencer_Egg = Void -> (Exports, (Imports, Run_Gun, End_Gun) -> Void); # PUBLIC.
Runstate = { # These values will be statically globally visible throughout the code body for the imp.
me: Xsequencer_Ximp_State, #
imports: Imports, # Ximps to which we send requests.
to: Replyqueue, # The name makes foo::pass_something(imp) to {. ... } syntax read well.
end_gun': End_Gun, # We shut down the microthread when this fires.
xpacket_q: Xpacket_Q, # Xpackets from inbuf_ximp -- src/lib/x-kit/xclient/src/wire/inbuf-ximp.pkg xerror_q: Xerror_Q, #
graphics_expose_event_accumulator: Ref (Null_Or( xet::x::Graphics_Expose_Record -> Void ) ) # Extra state for handling sequences of x::GRAPHICS_EXPOSE events.
};
Client_Q = Mailqueue( Runstate -> Void );
empty_v = v1u::from_list [];
# Convert "abc" -> "61.62.63." etc:
#
fun string_to_hex s
=
string::translate
(\\ c = number_string::pad_left '0' 2 (int::format number_string::HEX (char::to_int c)) + ".")
s;
# As above, starting with byte-vector:
#
fun bytes_to_hex bytes
=
string_to_hex (byte::unpack_string_vector(vector_slice_of_one_byte_unts::make_slice (bytes, 0, NULL)));
# Show printing chars verbatim, everything
# else as '.', per hexdump tradition:
#
fun string_to_ascii s
=
string::translate
(\\ c = char::is_print c ?? string::from_char c :: ".")
s;
# As above, starting with byte-vector:
#
fun bytes_to_ascii bytes
=
string_to_ascii (byte::unpack_string_vector (vector_slice_of_one_byte_unts::make_slice (bytes, 0, NULL)));
# +DEBUG
fun seqn_to_string n # "seqn" == "sequence number"
=
un::format number_string::DECIMAL n;
#
fun queue_element_to_string (ERROR_CHECK (n, _)) => " ERROR_CHECK seqn==" + (seqn_to_string n);
queue_element_to_string (ONE_REPLY (n, _)) => " ONE_REPLY seqn==" + (seqn_to_string n);
queue_element_to_string (EXPOSURE_REPLY (n, _)) => " EXPOSURE_REPLY seqn==" + (seqn_to_string n);
# queue_element_to_string (MULTI_REPLY (n, _, _, _)) => " MULTI_REPLY seqn==" + (seqn_to_string n);
end;
#
fun pending_reply_queue_to_string { front => [], rear => [] }
=>
"(pending reply queue is empty)";
pending_reply_queue_to_string { front, rear }
=>
"{" + (cat (queue_to_strings (front @ (reverse rear), []))) + "}"
where
fun queue_to_strings ([], l) => reverse l;
queue_to_strings (x ! r, l) => queue_to_strings (r, ((queue_element_to_string x) + "; ") ! l);
end;
end;
end;
# -DEBUG
#
fun seqn_of (ERROR_CHECK (seqn, _ )) => seqn;
seqn_of (ONE_REPLY (seqn, _ )) => seqn;
seqn_of (EXPOSURE_REPLY (seqn, _ )) => seqn;
# seqn_of (MULTI_REPLY (seqn, _, _, _)) => seqn;
end;
# Spawn throw-away thread to deliver
# multiple X server replies. This is
# to handle the currently-unused MULTI_REPLY:
#
# fun send_replies (slot, replies)
# =
# loop (reverse replies)
# where
# fun loop [] => ();
# #
# loop (s ! rest)
# =>
# { put_in_mailslot (slot, REPLY s);
# #
# loop rest;
# };
# end;
# end;
# Search pending-reply queue for the
# sequence number n, which is from the
# latest X server message received.
#
# If we have any pending replies with
# lower sequence numbers they must
# correspond to lost X server requests,
# so we do the best we can with them
# and then drop them from the queue.
#
# We return the pair
#
# { found_it, updated_queue }
#
# where:
#
# updated_queue
# is the updated queue.
#
# found_it
# is TRUE iff the head
# of updated_queue has
# sequence number n.
#
fun get_pending_reply_n (n, q)
=
get_pending_reply_n' q
where
fun handle_lost_reply (ERROR_CHECK(_, oneshot)) => put_in_oneshot (oneshot, x2s::REPLY empty_v);
handle_lost_reply (ONE_REPLY (_, oneshot)) => put_in_oneshot (oneshot, x2s::REPLY_LOST);
# handle_lost_reply (MULTI_REPLY(_, oneshot, _, [] )) => put_in_oneshot (oneshot, x2s::REPLY_LOST);
# handle_lost_reply (MULTI_REPLY(_, oneshot, _, replies)) => put_in_oneshot (oneshot, replies);
handle_lost_reply (EXPOSURE_REPLY(_, oneshot))
=>
put_in_oneshot (oneshot, \\ () = raise exception LOST_REPLY);
end;
#
fun get_pending_reply_n' (q' as { front => [], rear => [] })
=>
{ found_it => FALSE, updated_queue => q' };
get_pending_reply_n' { front => [], rear }
=>
get_pending_reply_n' { front => (reverse rear), rear => [] };
get_pending_reply_n' (q' as { front => pending_reply ! rest, rear })
=>
{ seqn = seqn_of pending_reply;
#
if (seqn < n)
#
handle_lost_reply pending_reply;
#
get_pending_reply_n' { front => rest, rear };
else
seqn > n ?? { found_it => FALSE, updated_queue => q' }
:: { found_it => TRUE, updated_queue => q' };
fi;
};
end;
end;
# Extract the pending-reply queue entry
# with the sequence number n.
#
# If all of the expected X server replies
# for that entry have been received then
# send the extracted reply to the requesting
# client.
#
fun handle_reply_message (seqn, reply, pending_reply_queue)
=
case (get_pending_reply_n (seqn, pending_reply_queue))
#
{ found_it => TRUE, updated_queue => { front => ONE_REPLY(_, oneshot) ! rest, rear } }
=>
{
put_in_oneshot (oneshot, x2s::REPLY reply);
{ front => rest, rear };
};
# { found_it => TRUE, updated_queue => { front => MULTI_REPLY (seqn, slot, remaining, replies) ! rest, rear } }
# =>
# if (remaining reply == 0)
# #
# send_replies (slot, reply ! replies);
# (rest, rear);
# else
# ( MULTI_REPLY (seqn, slot, remaining, reply ! replies) ! rest,
# rear
# );
# fi;
_ =>
{ # Debug support:
#
log::fatal
( sprintf "IMPOSSIBLE ERROR: xsocket::handle_reply_message(seqn==%s, reply x=%s (%d bytes)...): BOGUS PENDING REPLY QUEUE, queue =%s"
(seqn_to_string seqn)
(bytes_to_hex reply)
(v1u::length reply)
(pending_reply_queue_to_string pending_reply_queue)
); # Does not return
pending_reply_queue; # Cannot execute -- just to pacify compiler typechecker.
};
esac;
# Extract the pending-reply queue entry
# with seqence number n:
#
fun handle_expose_event_train (n, reply, pending_reply_queue)
=
{
case (get_pending_reply_n (n, pending_reply_queue))
#
{ found_it => TRUE,
updated_queue => { front => EXPOSURE_REPLY(_, oneshot) ! rest, rear }
}
=> {
put_in_oneshot (oneshot, \\ () = reply);
#
{ front => rest, rear };
};
# For now, just drop it.
# When the gc-server supports graphics-exposures,
# these shouldn't happen: XXX SUCKO FIXME
#
_ => {
pending_reply_queue;
};
esac;
# +DEBUG
# (dumpPendingQ (n, q);
# xgripe::impossible "ERROR: xsocket::handle_expose_event_train: bogus pending reply queue]")
# -DEBUG
};
# Extract the pending-reply queue entry
# with seqence number n (corresponding
# to the given error message):
#
fun handle_error_message (n, err, pending_reply_queue)
=
case (get_pending_reply_n (n, pending_reply_queue))
#
{ found_it => TRUE, updated_queue => { front => ERROR_CHECK(_, oneshot) ! rest, rear } }
=>
{ put_in_oneshot (oneshot, x2s::REPLY_ERROR err);
#
{ front => rest, rear };
};
{ found_it => TRUE, updated_queue => { front => ONE_REPLY(_, oneshot) ! rest, rear } }
=>
{ put_in_oneshot (oneshot, x2s::REPLY_ERROR err);
#
{ front => rest, rear };
};
# { found_it => TRUE, updated_queue => { front => MULTI_REPLY(_, oneshot, _, _) ! rest, rear } }
# =>
# { put_in_oneshot (oneshot, REPLY_ERROR err);
# { front => rest, rear };
# };
{ found_it => TRUE, updated_queue => { front => EXPOSURE_REPLY(_, oneshot) ! rest, rear } }
=>
{ put_in_oneshot (oneshot, \\ () = raise exception ERROR_REPLY (w2v::decode_error err));
#
{ front => rest, rear };
};
{ found_it => FALSE, updated_queue => pending_reply_queue' }
=>
pending_reply_queue';
_ =>
/* DEBUG */ { trace {. sprintf "IMPOSSIBLE ERROR: xsocket::handle_error_message(seqn==%s: BOGUS PENDING REPLY QUEUE, queue =%s" (seqn_to_string n) (pending_reply_queue_to_string pending_reply_queue); };
xgripe::impossible "ERROR: xsocket::handle_error_message: bogus pending reply queue]";
/* DEBUG */ };
esac;
#
fun handle_event_message (n, pending_reply_queue)
=
case (get_pending_reply_n (n, pending_reply_queue))
#
{ found_it => TRUE, updated_queue => { front => ERROR_CHECK(_, oneshot) ! rest, rear } }
=>
{ put_in_oneshot (oneshot, x2s::REPLY empty_v);
#
{ front => rest, rear };
};
{ found_it, updated_queue => pending_reply_queue' }
=>
pending_reply_queue';
esac;
fun run ( client_q: Client_Q, # Requests from x-widgets and such via draw_imp, pen_imp or font_imp.
#
runstate as
{ # These values will be statically globally visible throughout the code body for the imp.
me: Xsequencer_Ximp_State, #
imports: Imports, # Ximps to which we send requests.
to: Replyqueue, # The name makes foo::pass_something(imp) to {. ... } syntax read well.
end_gun': End_Gun, # We shut down the microthread when this fires.
xpacket_q: Xpacket_Q, # Xpackets from inbuf_ximp -- src/lib/x-kit/xclient/src/wire/inbuf-ximp.pkg xerror_q: Xerror_Q, #
graphics_expose_event_accumulator: Ref (Null_Or( xet::x::Graphics_Expose_Record -> Void ) ) # Extra state for handling sequences of x::GRAPHICS_EXPOSE events.
}
)
=
loop ()
where
fun loop () # Outer loop for the imp.
=
{ do_one_mailop' to [
#
end_gun' ==> shut_down_sequencer_imp',
take_all_from_mailqueue' client_q ==> do_client_pleas,
take_from_mailqueue' xpacket_q ==> do_xpacket_plea
];
loop ();
}
where
fun do_client_plea thunk
=
thunk runstate;
# Handle requests from clients
# (app threads on our side):
#
fun do_client_pleas [] => ();
#
do_client_pleas pleas
=>
{ apply do_client_plea pleas;
#
# put_in_mailslot (to_x_mailslot, FLUSH_OUTBUF);
};
end;
fun shut_down_sequencer_imp' ()
=
thread_exit { success => TRUE }; # Will not return.
#
# stipulate
# fun add_to_pending_reply_queue pending_reply
# =
# { (*me.pending_reply_queue) -> { front, rear };
# #
# me.pending_reply_queue := { front, rear => pending_reply ! rear };
# };
# #
# fun send_request_reply (request, reply_oneshot)
# =
# { n = *me.last_seqn_sent + 0u1;
# me.last_seqn_sent := n;
#
# imports.xsequencer_to_outbuf.put_value request;
#
# add_to_pending_reply_queue (ONE_REPLY (n, reply_oneshot));
# };
#
# #
# fun send_request_and_check (request, reply_oneshot)
# =
# { n = *me.last_seqn_sent + 0u1;
# me.last_seqn_sent := n;
# #
# imports.xsequencer_to_outbuf.put_value request;
#
# add_to_pending_reply_queue (ERROR_CHECK (n, reply_oneshot));
# };
#
# # fun send_request_replies (req, reply_mailslot, remain)
# # =
# # { n = *me.last_seqn_sent + 0u1;
# # me.last_seqn_sent := n;
# # #
# # put_in_mailslot (to_x_mailslot, ADD_TO_OUTBUF req);
# #
# # me.pending_reply_queue := add_to_pending_reply_queue (MULTI_REPLY (n, reply_mailslot, remain, []), *me.pending_reply_queue);
# # };
# #
# #
# fun do_clientplea (PLEA_REPLY request) => send_request_reply request;
# do_clientplea (PLEA_AND_CHECK request) => send_request_and_check request;
#
# # do_clientplea (PLEA_REPLIES request) => send_request_replies request;
# # #
# do_clientplea (PLEA_SEND_BYTEVECTOR request)
# =>
# { imports.xsequencer_to_outbuf.put_value request;
# #
# me.last_seqn_sent := *me.last_seqn_sent + 0u1;
# };
#
# do_clientplea (PLEA_SEND_BYTEVECTORS requests)
# =>
# { imports.xsequencer_to_outbuf.put_values requests;
# #
# me.last_seqn_sent := *me.last_seqn_sent + (unt::from_int (list::length requests));
# };
# end;
#
# herein
fun do_xpacket_plea (XPLEA_NOTE_XPACKET (xpacket as { code: v1u::Element, packet: v1u::Vector }))
=
{
fun get_seq_n () # Get sequence number for packet. Low 16 bits are in the packet, the rest we fill in.
=
{ short_seq_n = un::from_large_unt (pack_big_endian_unt16::get_vec (packet, 1));
#
seqn' = un::bitwise_or
( un::bitwise_and (*me.last_seqn_read, un::bitwise_not 0uxffff), # FFFF mask because X protocol reply packets contain only low 16 bits of the full sequence number.
short_seq_n
);
seqn' < *me.last_seqn_read
?? seqn' + 0ux10000 # NOTE: we should check for (seqn' + 0x10000) > lastReqOut XXX BUGGO FIXME
:: seqn';
};
#
# "NOTE: above logic doesn't work if there are 2**17
# outgoing messages between replies/events.
#
# "We need to track (last_seqn_sent - last_seqn_read),
# and if it gets bigger than some reasonable size,
# generate a synchronization (i.e., get_input_focus message)."
# -- John H Reppy
#
# But is there any reason to not just truncate these numbers to 16 bits? -- 2013-07-11 CrT
case code
#
0u0 => { # Error message:
#
seqn = get_seq_n ();
me.last_seqn_read := seqn;
put_in_mailqueue (xerror_q, { seqn, msg => packet });
me.pending_reply_queue := handle_error_message (seqn, packet, *me.pending_reply_queue);
};
0u1 => { # Reply message:
#
seqn = get_seq_n ();
me.last_seqn_read := seqn;
me.pending_reply_queue := handle_reply_message (seqn, packet, *me.pending_reply_queue);
};
0u11 => { # KeymapNotify event:
#
imports.xpacket_sink.put_value { code, packet };
me.pending_reply_queue := handle_event_message (*me.last_seqn_read, *me.pending_reply_queue);
};
0u13 => { # GraphicsExpose event:
#
seqn = get_seq_n ();
me.last_seqn_read := seqn;
include package xet; # xevent_types is from src/lib/x-kit/xclient/src/wire/xevent-types.pkg graphics_expose_record
=
case (w2v::decode_graphics_expose packet)
#
xet::x::GRAPHICS_EXPOSE graphics_expose_record => graphics_expose_record;
_ => raise exception DIE "Impossible case";
esac;
case *graphics_expose_event_accumulator
#
NULL => accumulate_graphics_expose_events [] graphics_expose_record; # Start accumulating GRAPHICS_EXPOSE events in a fresh sequence.
THE accumulator => accumulator graphics_expose_record; # Continue accumulating GRAPHICS_EXPOSE events in existing sequence.
esac
where
# The X server sends numbered trains of expose events.
# We use our 'graphics_expose_event_accumulator' refcell to accumulate
# a train of expose events, then handle it when complete:
#
fun accumulate_graphics_expose_events boxes ({ box, count=>0, ... }: xet::x::Graphics_Expose_Record) # Note currying.
=>
{ me.pending_reply_queue := handle_expose_event_train (seqn, box ! boxes, *me.pending_reply_queue); # Sequence complete -- pass boxes to client code.
#
graphics_expose_event_accumulator := NULL; # Done with this expose event sequence.
};
accumulate_graphics_expose_events boxes ({ box, ... }: xet::x::Graphics_Expose_Record) # Sequence not complete -- continue accumulation.
=>
{
graphics_expose_event_accumulator := THE (accumulate_graphics_expose_events (box ! boxes)); # Note partial application of curried fn.
};
end;
end;
};
0u14 => { # NoExpose event:
#
seqn = get_seq_n ();
me.last_seqn_read := seqn;
me.pending_reply_queue := handle_expose_event_train (seqn, [], *me.pending_reply_queue);
};
_ => { # Other event packets:
#
seqn = get_seq_n ();
me.last_seqn_read := seqn;
imports.xpacket_sink.put_value { code, packet };
me.pending_reply_queue := handle_event_message (seqn, *me.pending_reply_queue);
};
esac;
};
# end;
end; # fun loop
end; # fun run
fun startup (reply_oneshot: Oneshot_Maildrop( (Me_Slot, Exports) )) () # Root fn of imp microthread. Note currying.
=
{ me_slot = make_mailslot () : Me_Slot;
#
xpacket_sink = { put_value };
xclient_to_sequencer
=
{ send_xrequest,
send_xrequests,
send_xrequest_and_read_reply,
send_xrequest_and_read_reply',
send_xrequest_and_pass_reply,
send_xrequest_and_return_completion_mailop,
send_xrequest_and_return_completion_mailop'
};
xerror_well = { take_xerror,
take_xerror'
};
to = make_replyqueue();
put_in_oneshot (reply_oneshot, (me_slot, { xpacket_sink, xclient_to_sequencer, xerror_well })); # Return value from xsequencer_egg'().
(take_from_mailslot me_slot) # Imports from xsequencer_egg'().
->
{ me, imports, run_gun', end_gun' };
block_until_mailop_fires run_gun'; # Wait for the starting gun.
graphics_expose_event_accumulator = REF NULL;
run (client_q, { me, xpacket_q, xerror_q, imports, to, end_gun', graphics_expose_event_accumulator }); # Will not return.
}
where
xpacket_q = make_mailqueue (get_current_microthread()) : Xpacket_Q;
client_q = make_mailqueue (get_current_microthread()) : Client_Q;
xerror_q = make_mailqueue (get_current_microthread()) : Xerror_Q;
# Reply handling in the Client-thread context.
#
# Most processing happens in our own microthread,
# but any client-relevant exception
# needs to be raised in the context of the
# calling client thread. That is our job here:
#
fun unwrap_reply x2s::REPLY_LOST => { log::fatal "xsequencer-ximp.pkg: Lost X-server reply"; raise exception DIE "LOST REPLY"; };
unwrap_reply (x2s::REPLY_ERROR s) => { log::fatal ( "xsequencer-ximp.pkg: X-server error: " + (e2s::xerror_to_string (w2v::decode_error s))); raise exception DIE "ERROR_REPLY"; };
unwrap_reply (x2s::REPLY s) => s; # NB log::fatal should never return;
end; # above 'raises' keep typechecker happy.
fun unwrap_flag r
=
{ unwrap_reply r;
();
};
#
fun add_to_pending_reply_queue
(
me: Xsequencer_Ximp_State,
pending_reply: Pending_Reply
)
=
{ (*me.pending_reply_queue) -> { front, rear };
#
me.pending_reply_queue := { front, rear => pending_reply ! rear };
};
#
fun send_xrequest (request: v1u::Vector) # PUBLIC.
=
{
put_in_mailqueue (client_q,
#
\\ ({ me, imports, ... }: Runstate)
=
{ imports.xsequencer_to_outbuf.put_value request;
#
me.last_seqn_sent := *me.last_seqn_sent + 0u1;
}
);
};
#
fun send_xrequests (requests: List(v1u::Vector)) # PUBLIC.
=
{
put_in_mailqueue (client_q,
#
\\ ({ me, imports, ... }: Runstate)
=
{ imports.xsequencer_to_outbuf.put_values requests;
#
me.last_seqn_sent := *me.last_seqn_sent + (unt::from_int (list::length requests));
}
);
};
# This is a workhorse call,
# request-with-single-reply:
#
fun send_xrequest_and_read_reply (request: v1u::Vector) # PUBLIC.
=
{ reply_oneshot = make_oneshot_maildrop ();
#
put_in_mailqueue (client_q,
#
\\ ({ me, imports, ... }: Runstate)
=
{ n = *me.last_seqn_sent + 0u1;
me.last_seqn_sent := n;
imports.xsequencer_to_outbuf.put_value request;
add_to_pending_reply_queue (me, ONE_REPLY (n, reply_oneshot));
}
);
get_from_oneshot' reply_oneshot
==>
unwrap_reply;
};
# As above, but with client providing the oneshot:
#
fun send_xrequest_and_read_reply' # PUBLIC.
(
request: v1u::Vector,
reply_oneshot: Oneshot_Maildrop(x2s::Reply_Mail)
)
=
put_in_mailqueue (client_q,
#
\\ ({ me, imports, ... }: Runstate)
=
{ n = *me.last_seqn_sent + 0u1;
me.last_seqn_sent := n;
imports.xsequencer_to_outbuf.put_value request;
add_to_pending_reply_queue (me, ONE_REPLY (n, reply_oneshot));
}
);
fun send_xrequest_and_pass_reply # PUBLIC.
(request: v1u::Vector)
(replyqueue: Replyqueue)
(reply_handler: v1u::Vector -> Void)
=
{ reply_oneshot = make_oneshot_maildrop(): Oneshot_Maildrop( x2s::Reply_Mail );
#
put_in_mailqueue (client_q,
#
\\ ({ me, imports, ... }: Runstate)
=
{ n = *me.last_seqn_sent + 0u1;
me.last_seqn_sent := n;
imports.xsequencer_to_outbuf.put_value request;
add_to_pending_reply_queue (me, ONE_REPLY (n, reply_oneshot));
}
);
put_in_replyqueue (replyqueue, (get_from_oneshot' reply_oneshot) ==> (reply_handler o unwrap_reply));
};
# Generate a request to the server and
# check on its successful completion.
#
# The only uses of this I find are:
#
# property::change_property in
# src/lib/x-kit/xclient/src/iccc/window-property-old.pkg #
# font_imp::open_font in
# src/lib/x-kit/xclient/src/window/font-imp-old.pkg #
# In both cases the idea is to wait for
# successful completion of the op before
# continuing.
#
fun send_xrequest_and_return_completion_mailop (request1: v1u::Vector) # PUBLIC.
=
{ reply_oneshot1 = make_oneshot_maildrop (); # Nobody will read this.
reply_oneshot2 = make_oneshot_maildrop (): Oneshot_Maildrop(x2s::Reply_Mail);
put_in_mailqueue (client_q,
#
\\ ({ me, imports, ... }: Runstate)
=
{ n = *me.last_seqn_sent + 0u1;
me.last_seqn_sent := n;
#
imports.xsequencer_to_outbuf.put_value request1;
add_to_pending_reply_queue (me, ERROR_CHECK (n, reply_oneshot1));
}
);
# Reppy often uses dummy get_input_focus calls, presumably
# to verify that preceding commands have completed.
put_in_mailqueue (client_q,
#
\\ ({ me, imports, ... }: Runstate)
=
{ n = *me.last_seqn_sent + 0u1;
me.last_seqn_sent := n;
imports.xsequencer_to_outbuf.put_value v2w::request_get_input_focus;
add_to_pending_reply_queue (me, ONE_REPLY (n, reply_oneshot2));
}
);
# Construct and return a mailop which caller can
# block_until_mailop_fires
# on to await completion of the requested operation:
#
get_from_oneshot' reply_oneshot2 # This was using 'reply_oneshot1' which made no sense to me, so I changed it to
==> # 'reply_oneshot2' on the presumption that it was a coding error. -- 2013-07-04 CrT
unwrap_flag;
};
# As above, but designed for chaining from xserver-ximp.pkg:
#
fun send_xrequest_and_return_completion_mailop' # PUBLIC.
(
request1: v1u::Vector,
reply_oneshot2: Oneshot_Maildrop(x2s::Reply_Mail)
)
=
{ reply_oneshot1 = make_oneshot_maildrop (); # Nobody will read this.
#
put_in_mailqueue (client_q,
#
\\ ({ me, imports, ... }: Runstate)
=
{ n = *me.last_seqn_sent + 0u1;
me.last_seqn_sent := n;
#
imports.xsequencer_to_outbuf.put_value request1;
add_to_pending_reply_queue (me, ERROR_CHECK (n, reply_oneshot1));
}
);
put_in_mailqueue (client_q,
#
\\ ({ me, imports, ... }: Runstate)
=
{ n = *me.last_seqn_sent + 0u1;
me.last_seqn_sent := n;
imports.xsequencer_to_outbuf.put_value v2w::request_get_input_focus;
add_to_pending_reply_queue (me, ONE_REPLY (n, reply_oneshot2));
}
);
};
take_xerror = \\ () = take_from_mailqueue xerror_q;
take_xerror' = take_from_mailqueue' xerror_q;
#
fun put_value (xpacket: xps::Xpacket) # PUBLIC. inbuf-ximp calls this to pass us a packet from xserver.
=
{
put_in_mailqueue (xpacket_q, XPLEA_NOTE_XPACKET xpacket);
};
end;
fun process_options (options: List(Option), { name })
=
{ my_name = REF name;
#
apply do_option options
where
fun do_option (MICROTHREAD_NAME n) = my_name := n;
end;
{ name => *my_name };
};
##########################################################################################
# PUBLIC.
#
fun make_xsequencer_egg (options: List(Option)) # PUBLIC. PHASE 1: Construct our state and initialize from 'options'.
=
{ (process_options (options, { name => "tmp" }))
->
{ name };
me = { last_seqn_read => REF 0u0,
last_seqn_sent => REF 0u0,
pending_reply_queue => REF { front => [],
rear => []
}
};
\\ () = { reply_oneshot = make_oneshot_maildrop(): Oneshot_Maildrop( (Me_Slot, Exports) ); # PUBLIC. PHASE 2: Start our microthread and return our Exports to caller.
#
xlogger::make_thread name (startup reply_oneshot); # Note that startup() is curried.
(get_from_oneshot reply_oneshot) -> (me_slot, exports);
fun phase3 # PUBLIC. PHASE 3: Accept our Imports, then wait for Run_Gun to fire.
(
imports: Imports,
run_gun': Run_Gun,
end_gun': End_Gun
)
=
{
put_in_mailslot (me_slot, { me, imports, run_gun', end_gun' });
};
(exports, phase3);
};
};
}; # package xsequencer_ximp
end;