"Cleaning" the message queue

classic Classic list List threaded Threaded
2 messages Options
Reply | Threaded
Open this post in threaded view
|

"Cleaning" the message queue

Vlad Dumitrescu-3
Hi all,

I have an application where processes get messages posted quite often, in a
periodical fashion. Sometimes they don't have time to finish processing one
message before another drops in, so the message queue grows longer and
longer.

What I'd want is that in case there are several such messages waiting, only
the latest shall be processed, while the rest shall be dumped. (The messages
are "virtual clock ticks")

I tried some schemes of setting up things, but none worked. It may be
supposed that there will only be one kind of messages in the queue.

What am I missing? Maybe there is another way to assure many processes get
common virtual time information? Where should I look?

Thanks in advance,
Vlad
_________________________________________________________________________
Get Your Private, Free E-mail from MSN Hotmail at http://www.hotmail.com.



Reply | Threaded
Open this post in threaded view
|

"Cleaning" the message queue

Luke Gorrie-3
"Vlad Dumitrescu" <vladdu> writes:

> Hi all,
>
> I have an application where processes get messages posted quite often,
> in a periodical fashion. Sometimes they don't have time to finish
> processing one message before another drops in, so the message queue
> grows longer and longer.
>
> What I'd want is that in case there are several such messages waiting,
> only the latest shall be processed, while the rest shall be
> dumped. (The messages are "virtual clock ticks")

Here's an example program with a "receiver" process that gets 'go'
messages, and discards duplicates that pile up on its message
queue. Hopefully its pretty similar to your situation.

Since I have no information in my "go" messages I'm not actually using
the "latest" one, but that'd be an easy change.

--- SNIP ---

-module(receive_one).

-export([test/0]).

-export([receiver/0]).                  % internal

%% Test: We create a "receiver" process that will handle some 'go'
%% messages. We expect it to only actually handle two of them and
%% discard the rest, due to the timings!
%%
%% So we expect the string "Going!" should be io:format'd just twice.
test() ->
    Receiver = spawn_link(?MODULE, receiver, []),
    Receiver ! go,
    %% After this wait, the receiver will be busy processing the first
    %% 'go', allowing the next ones to accumulate in its message queue
    wait(5),
    Receiver ! go,
    Receiver ! go,
    Receiver ! go,
    Receiver ! go.

receiver() ->
    receive_go(),
    process(),
    receiver().

%% Process a 'go' message. This takes some time.
process() ->
    io:format("Going!~n"),
    %% We spend 500ms "processing" each message, so that there's time
    %% for new ones to accumulate on our message queue.
    wait(500).

%% Wait to receive a 'go' message, and flush any extra ones.
receive_go() ->
    receive
        go ->
            flush_go_msgs(),
            ok
    end.

flush_go_msgs() ->
    %% Receive/discard go messages until we timeout with 'after 0' -
    %% i.e. until no go messages are left on message queue
    receive
        go -> flush_go_msgs()
    after
        0 -> no_more_gos
    end.

wait(Time) ->
    receive
        i_never_get_this -> exit(impossible)
    after
        Time -> ok
    end.