cosnotification_eventdb.erl

来自「OTP是开放电信平台的简称」· ERL 代码 · 共 1,350 行 · 第 1/4 页

%%--------------------------------------------------------------------
%% ``The contents of this file are subject to the Erlang Public License,
%% Version 1.1, (the "License"); you may not use this file except in
%% compliance with the License. You should have received a copy of the
%% Erlang Public License along with this software. If not, it can be
%% retrieved via the world wide web at http://www.erlang.org/.
%% 
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
%% the License for the specific language governing rights and limitations
%% under the License.
%% 
%% The Initial Developer of the Original Code is Ericsson Utvecklings AB.
%% Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
%% AB. All Rights Reserved.''
%% 
%%     $Id$
%%
%%----------------------------------------------------------------------
%% File    : cosNotification_eventDB.erl
%% Purpose : 
%% Created : 24 Mar 2000
%% Purpose : This module is supposed to centralize Event storage.
%% Comments: 
%% * Setting Order Policy to AnyOrder eq. Priority order
%%
%% * Setting Discard Policy to AnyOrder eq. RejectNewEvents.
%%
%% * DB ordering: Since the deliver- and discard-order may differ we need
%%   two ets-tables, both of type 'ordered_set'. They contain:
%%   - table 1 (T1): deliver order key and the associated event.
%%   - table 2 (T2): discard order key.
%%   
%%   When adding a new event we add, if necessary, related keys in T2.
%%   For example, if we should discard events in FIFO order, the delivery
%%   order may be set to Priority order. If the Max Event limit is reached
%%   we first look in T2 to find out which event to discard by using and
%%   reorder the key elements. T2 gives {TimeStamp, Priority}, which is used
%%   to lookup in T1 as {Priority, TimeStamp}.
%%   A TimeStamp is always included in the DB keys, even if FIFO or LIFO
%%   is used, since lots of events probably will have the same prioity and
%%   with a little bit of bad luck some events will never be delivered.
%%   
%%   Note: deliver order AnyOrder and PriorityOrder is equal since the later
%%         is defined as default.
%%         discard order AnyOrder and RejectNewEvents is equal since the later
%%         is defined as default.
%%   The keys used is ('-' indicates T2 is not needed and, thus, not instantiated):
%%   
%%   T1 policy         T1 Key             T2 Policy       T2 Key
%%   ------------------------------------------------------------------
%%   DeadlineOrder     {DL, Key, Prio}    PriorityOrder   {Prio, Key, DL}
%%   DeadlineOrder     {DL, Key}          FifoOrder       {Key, DL}
%%   DeadlineOrder     {DL, Key}          LifoOrder       {Key, DL}
%%   DeadlineOrder     {DL, Key}          RejectNewEvents     -
%%   DeadlineOrder     {DL, Key}          DeadlineOrder       -
%%   FifoOrder         {Key, DL}          DeadlineOrder   {DL, Key}
%%   FifoOrder         {Key, Prio}        PriorityOrder   {Prio, Key}
%%   FifoOrder         Key                RejectNewEvents     -
%%   FifoOrder         Key                Fifo                -
%%   FifoOrder         Key                Lifo                -
%%   PriorityOrder     {Prio, Key, DL}    DeadlineOrder   {DL, Key, Prio}
%%   PriorityOrder     {Prio, Key}        Fifo            {Key, Prio}
%%   PriorityOrder     {Prio, Key}        Lifo            {Key, Prio}
%%   PriorityOrder     {Prio, Key}        RejectNewEvents     -
%%   ------------------------------------------------------------------
%%   DL == Deadline, Key == TimeStamp, Prio == Priority
%%   
%%   NOTE: If defined, the Discard DB Keys are the same as in Event DB, except 
%%   that the first and last Key change place. {K1,K2}<->{K2,K1} and 
%%   {K1,K2,K3}<->{K3,K2,K1}.
%%----------------------------------------------------------------------

-module(cosNotification_eventDB).


%%--------------- INCLUDES -----------------------------------
-include_lib("orber/include/corba.hrl").
-include_lib("orber/include/ifr_types.hrl").
-include_lib("cosTime/include/TimeBase.hrl").

%% Application files
-include("CosNotification.hrl").
-include("CosNotifyChannelAdmin.hrl").
-include("CosNotifyComm.hrl").
-include("CosNotifyFilter.hrl").

-include("CosNotification_Definitions.hrl").

%%--------------- EXPORTS ------------------------------------
%% Internal Filter Functions
-export([validate_event/5, 
	 create_db/4,
	 destroy_db/1,
	 get_event/1,
	 get_event/2,
	 get_events/2,
	 get_events/3,
	 delete_events/1,
	 update/2,
	 update/4,
	 add_event/2,
	 add_event/4,
	 add_and_get_event/2,
	 add_and_get_event/3,
	 add_and_get_event/4,
	 add_and_get_event/5,
	 gc_events/2,
	 gc_events_local/4,
	 gc_start/2,
	 filter_events/2,
	 filter_events/3,
	 status/2]).

%%--------------- DATA STRUCTURES ----------------------------
-record(dbRef, {orderRef, discardRef, orderPolicy, discardPolicy,
		defPriority, maxEvents, defStopT, startTsupport,
		stopTsupport, gcTime, gcLimit, timeRef}).



%%--------------- DEFINES ------------------------------------

-define(CreateRef(OR, DR, O, D, DP, ME, DS, StaT, StoT, GT, GL, TR), 
	#dbRef{orderRef=OR, discardRef=DR, orderPolicy=O, discardPolicy=D,
	       defPriority=DP, maxEvents=ME, defStopT=DS, startTsupport=StaT,
	       stopTsupport=StoT, gcTime=GT, gcLimit=round(ME*GL/100),
	       timeRef=TR}).


-define(get_OrderP(DR),            DR#dbRef.orderPolicy).
-define(get_DiscardP(DR),          DR#dbRef.discardPolicy).
-define(get_OrderRef(DR),          DR#dbRef.orderRef).
-define(get_DiscardRef(DR),        DR#dbRef.orderRef).
-define(get_DefPriority(DR),       DR#dbRef.defPriority).
-define(get_MaxEvents(DR),         DR#dbRef.maxEvents).
-define(get_DefStopT(DR),          DR#dbRef.defStopT).
-define(get_StartTsupport(DR),     DR#dbRef.startTsupport).
-define(get_StopTsupport(DR),      DR#dbRef.stopTsupport).
-define(get_GCTime(DR),            DR#dbRef.gcTime).
-define(get_GCLimit(DR),           DR#dbRef.gcLimit).
-define(get_TimeRef(DR),           DR#dbRef.timeRef).

-define(set_OrderP(DR, O),         DR#dbRef{orderPolicy = O}).
-define(set_DiscardP(DR, D),       DR#dbRef{discardPolicy = D}).
-define(set_OrderRef(DR, E),       DR#dbRef{orderRef = E}).
-define(set_DiscardRef(DR, E),     DR#dbRef{orderRef = E}).
-define(set_DefPriority(DR, DP),   DR#dbRef{defPriority = DP}).
-define(set_MaxEvents(DR, ME),     DR#dbRef{maxEvents = ME}).
-define(set_DefStopT(DR, DS),      DR#dbRef{defStopT = DS}).
-define(set_StartTsupport(DR, B),  DR#dbRef{startTsupport = B}).
-define(set_StopTsupport(DR, B),   DR#dbRef{stopTsupport = B}).

-define(is_StartTNotSupported(DR), DR#dbRef.startTsupport == false).
-define(is_StopTNotSupported(DR),  DR#dbRef.stopTsupport  == false).
-define(is_TimeoutNotUsed(DR),     DR#dbRef.defStopT  == 0).


%%------------------------------------------------------------
%% function : status
%% Arguments: DBRef
%%            Key - which information we want.
%% Returns  : Data related to the Key.
%%------------------------------------------------------------
status(DBRef, eventCounter) ->
    ets:info(?get_OrderRef(DBRef), size);
status(DBRef, {batchLimit, Limit}) ->
    case ets:info(?get_OrderRef(DBRef), size) of
	Current when integer(Current), Current >= Limit ->
	    ?debug_print("BATCH LIMIT (~p) REACHED, CONTAINS: ~p~n", [Limit, Current]),
	    true;
	_Other ->
	    ?debug_print("BATCH LIMIT (~p) NOT REACHED, CONTAINS: ~p~n", 
			 [Limit, _Other]),
	    false
    end;
status(DBRef, {batchLimit, Limit, TemporaryMax}) ->
    case ets:info(?get_OrderRef(DBRef), size) of
	Current when integer(Current), Current >= TemporaryMax ->
	    ?debug_print("MAX LIMIT (~p) REACHED, CONTAINS: ~p~n", 
			 [TemporaryMax, Current]),
	    true;
	Current when integer(Current), Current >= Limit ->
	    ?debug_print("BATCH LIMIT (~p) REACHED, CONTAINS: ~p~n", [Limit, Current]),
	    true;
	_Other ->
	    ?debug_print("BATCH LIMIT (~p) NOT REACHED, CONTAINS: ~p~n", 
			 [Limit, _Other]),
	    false
    end;
status(_, _) ->
    error.


%%------------------------------------------------------------
%% function : gc_events_local
%% Arguments: DBRef
%% Returns  : 
%% Comment  : This function is intended for "emergency" GC, i.e.,
%%            when the DB must discard events we should first try
%%            to remove events with expired deadlines.
%%------------------------------------------------------------
gc_events_local(_, _, false, _) ->
    ok;
gc_events_local(_, _, _, 0) ->
    ok;
gc_events_local(ORef, DRef, _, _) ->
    gc_loop(ets:first(ORef), ORef, DRef).

%%------------------------------------------------------------
%% function : gc_events
%% Arguments: DBRef
%%            Priority - 'low', 'medium' or 'high'; will determine
%%            how important a fast gc is.
%% Returns  : 
%% Comment  : This function is intended for background GC.
%%------------------------------------------------------------
gc_events(DBRef, _Priority) when ?is_TimeoutNotUsed(DBRef) ->
    ok;
gc_events(DBRef, _Priority) when ?is_StopTNotSupported(DBRef) ->
    ok;
gc_events(DBRef, Priority) ->
    {M,S,U} = now(),
    case get(oe_GC_timestamp) of
	Num when {M,S,U} > Num ->
	    put(oe_GC_timestamp, {M,S+?get_GCTime(DBRef),U}),
	    spawn_link(?MODULE, gc_start, [DBRef, Priority]);
	_->
	    ok
    end.


%%------------------------------------------------------------
%% function : gc_start
%% Arguments: 
%% Returns  : 
%%------------------------------------------------------------
gc_start(#dbRef{orderRef = ORef, discardRef = DRef}, Priority) ->
    process_flag(priority, Priority),
    gc_loop(ets:first(ORef), ORef, DRef).

gc_loop('$end_of_table', _, _) ->
    ok;
gc_loop(Key, ORef, DRef) ->
    [{Keys,DL,_,_,_}]=ets:lookup(ORef, Key),
    case check_deadline(DL) of
	true when DRef == undefined ->
	    ets:delete(ORef, Key);
	true ->
	    ets:delete(ORef, Key),
	    gc_discard_DB(Keys, DRef);
	_ ->
	    ok
    end,
    gc_loop(ets:next(ORef, Key), ORef, DRef).

gc_discard_DB({Key1, Key2}, DRef) ->
    ets:delete(DRef, {Key2, Key1});
gc_discard_DB({Key1, Key2, Key3}, DRef) ->
    ets:delete(DRef, {Key3, Key2, Key1}).

%%------------------------------------------------------------
%% function : create_FIFO_Key
%% Arguments: 
%% Returns  : 
%%------------------------------------------------------------
create_FIFO_Key() ->
    {M, S, U} = erlang:now(),
    -M*1000000000000 - S*1000000 - U.

%%------------------------------------------------------------
%% function : convert_FIFO_Key
%% Arguments: 
%% Returns  : A now tuple
%% Comment  : Used when we must reuse a timestamp, i.e., only
%%            when we must reorder the DB.
%%------------------------------------------------------------
convert_FIFO_Key(Key) ->
    K = abs(Key),
    Secs = trunc(K/1000000),
    M = trunc(K/1000000000000),
    S = Secs-M*1000000,
    U = K - S*1000000-M*1000000000000,
    {M, S, U}.

%%------------------------------------------------------------
%% function : extract_priority
%% Arguments: Event
%%            Defalt Value
%%            Mapping Filter Value 
%%             - false  value not needed (depends on QoS type)
%%             - undefined value needed but no filter associated.
%% Returns  : 
%%------------------------------------------------------------
extract_priority(_, _, false) ->
    false;
extract_priority(#'CosNotification_StructuredEvent'
		 {header = #'CosNotification_EventHeader'
		  {variable_header = VH}}, DefPriority, undefined) ->
    extract_value(VH, ?not_Priority, DefPriority);
%% Maybe a unstructured event.
extract_priority(_, DefPriority, undefined) ->
    DefPriority;
extract_priority(_, _, PriorityOverride) ->
    %% Must have an associated MappingFilter for Priority.
    PriorityOverride.

%%------------------------------------------------------------
%% function : extract_start_time
%% Arguments: 
%% Returns  : 
%%------------------------------------------------------------
extract_start_time(_, false, _) ->
    false;
extract_start_time(#'CosNotification_StructuredEvent'
		 {header = #'CosNotification_EventHeader'
		  {variable_header = VH}}, _, TRef) ->
    ST = case extract_value(VH, ?not_StartTime, undefined) of
	     UTC when record(UTC, 'TimeBase_UtcT') ->
		 UTC;
	     _ ->
		 false
	 end,
    convert_time(ST, TRef, now());
extract_start_time(_, _, _) ->
    false.

%%------------------------------------------------------------
%% function : extract_deadline
%% Arguments: Structured Event
%%            Default Timeout Value - TimeT or UtcT (see cosTime).
%%            StopTSupported - boolean().
%%            TRef - reference to TimeService
%%            Mapping Filter Value 
%%             - false eq. value not needed (depends on QoS type)
%%             - undefined eq. value needed but no filter associated.
%%            Now - used when we want to reuse old TimeStamp which
%%                  must be done when changing QoS.