Skip to content

Commit

Permalink
To change the backend-db from bitcask to leveldb#1 for fixing leo-pro…
Browse files Browse the repository at this point in the history 
…ject/leofs/issues/437
  • Loading branch information
@yosukehara
yosukehara committed Dec 30, 2015
1 parent fe664d4 commit 730a6e2
Show file tree
Hide file tree
Showing 3 changed files with 163 additions and 161 deletions.
135 changes: 67 additions & 68 deletions include/leo_mq.hrl
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,109 +23,108 @@
%% @doc
%% @end
%%======================================================================
-define(MQ_LOG_ID, 'queue').
-define(MQ_LOG_FILE_NAME, "queue").
-define(MQ_PROP_MOD, 'module').
-define(MQ_PROP_FUN, 'function').
-define(MQ_PROP_DB_NAME, 'db_name').
-define(MQ_PROP_DB_PROCS, 'db_procs').
-define(MQ_PROP_ROOT_PATH, 'root_path').
-define(MQ_LOG_ID, 'queue').
-define(MQ_LOG_FILE_NAME, "queue").
-define(MQ_PROP_MOD, 'module').
-define(MQ_PROP_FUN, 'function').
-define(MQ_PROP_DB_NAME, 'db_name').
-define(MQ_PROP_DB_PROCS, 'db_procs').
-define(MQ_PROP_ROOT_PATH, 'root_path').
-define(MQ_PROP_NUM_OF_BATCH_PROC, 'num_of_batch_processes').
-define(MQ_SUBSCRIBE_FUN, 'subscribe').
-define(MQ_SUBSCRIBE_FUN, 'subscribe').

-define(MQ_PROP_INTERVAL_MAX, 'interval_max').
-define(MQ_PROP_INTERVAL_REG, 'interval_reg').
-define(MQ_PROP_BATCH_MSGS_MAX, 'batch_of_msgs_max').
-define(MQ_PROP_BATCH_MSGS_REG, 'batch_of_msgs_reg').
-define(MQ_PROP_NUM_OF_STEPS, 'num_of_steps').
-define(MQ_PROP_INTERVAL_MAX, 'interval_max').
-define(MQ_PROP_INTERVAL_REG, 'interval_reg').
-define(MQ_PROP_BATCH_MSGS_MAX, 'batch_of_msgs_max').
-define(MQ_PROP_BATCH_MSGS_REG, 'batch_of_msgs_reg').
-define(MQ_PROP_NUM_OF_STEPS, 'num_of_steps').

-define(DEF_BACKEND_DB_PROCS, 3).
-define(DEF_BACKEND_DB, 'bitcask').
-define(DEF_BACKEND_DB, 'leveldb').
%% -define(DEF_BACKEND_DB, 'bitcask').
-define(DEF_DB_ROOT_PATH, "mq" ).

-ifdef(TEST).
-define(DEF_CONSUME_MAX_INTERVAL, 1000).
-define(DEF_CONSUME_MIN_INTERVAL, 100).
-define(DEF_CONSUME_REG_INTERVAL, 300).
-define(DEF_CONSUME_MAX_BATCH_MSGS, 10).
-define(DEF_CONSUME_MIN_BATCH_MSGS, 1).
-define(DEF_CONSUME_REG_BATCH_MSGS, 5).
-define(DEF_CONSUME_NUM_OF_STEPS, 5).
-define(DEF_CONSUME_MAX_INTERVAL, 1000).
-define(DEF_CONSUME_MIN_INTERVAL, 100).
-define(DEF_CONSUME_REG_INTERVAL, 300).
-define(DEF_CONSUME_MAX_BATCH_MSGS, 10).
-define(DEF_CONSUME_MIN_BATCH_MSGS, 1).
-define(DEF_CONSUME_REG_BATCH_MSGS, 5).
-define(DEF_CONSUME_NUM_OF_STEPS, 5).
-else.
-define(DEF_CONSUME_MAX_INTERVAL, 3000).
-define(DEF_CONSUME_MIN_INTERVAL, 100).
-define(DEF_CONSUME_REG_INTERVAL, 300).
-define(DEF_CONSUME_MAX_BATCH_MSGS, 1000).
-define(DEF_CONSUME_MIN_BATCH_MSGS, 100).
-define(DEF_CONSUME_REG_BATCH_MSGS, 300).
-define(DEF_CONSUME_NUM_OF_STEPS, 10).
-define(DEF_CONSUME_MAX_INTERVAL, 3000).
-define(DEF_CONSUME_MIN_INTERVAL, 100).
-define(DEF_CONSUME_REG_INTERVAL, 300).
-define(DEF_CONSUME_MAX_BATCH_MSGS, 1000).
-define(DEF_CONSUME_MIN_BATCH_MSGS, 100).
-define(DEF_CONSUME_REG_BATCH_MSGS, 300).
-define(DEF_CONSUME_NUM_OF_STEPS, 10).
-endif.

-define(MQ_CNS_PROP_NUM_OF_MSGS, 'consumer_num_of_msgs').
-define(MQ_CNS_PROP_STATUS, 'consumer_status').
-define(MQ_CNS_PROP_NUM_OF_MSGS, 'consumer_num_of_msgs').
-define(MQ_CNS_PROP_STATUS, 'consumer_status').
-define(MQ_CNS_PROP_BATCH_OF_MSGS, 'consumer_batch_of_msgs').
-define(MQ_CNS_PROP_INTERVAL, 'consumer_interval').
-define(MQ_CNS_PROP_INTERVAL, 'consumer_interval').

-record(mq_properties, {
publisher_id :: atom(), %% publisher-id
consumer_id :: atom(), %% consumer-id
mod_callback :: module(), %% callback module name
db_name :: atom(), %% db's id
db_procs = 1 :: integer(), %% db's processes
root_path = [] :: string(), %% db's path
mqdb_id :: atom(), %% mqdb's id
mqdb_path = [] :: string(), %% mqdb's path
publisher_id :: atom(), %% publisher-id
consumer_id :: atom(), %% consumer-id
mod_callback :: module(), %% callback module name
db_name :: atom(), %% db's id
db_procs = 1 :: integer(), %% db's processes
root_path = [] :: string(), %% db's path
mqdb_id :: atom(), %% mqdb's id
mqdb_path = [] :: string(), %% mqdb's path
%% interval between batch-procs
max_interval = 1000 :: pos_integer(), %% max waiting time (default: 1000msec (1sec))
max_interval = 1000 :: pos_integer(), %% max waiting time (default: 1000msec (1sec))
regular_interval = 300 :: pos_integer(), %% regular waiting time (default: 300msec)
%% num of batch procs
max_batch_of_msgs = 1000 :: pos_integer(), %% max num of batch of messages
max_batch_of_msgs = 1000 :: pos_integer(), %% max num of batch of messages
regular_batch_of_msgs = 300 :: pos_integer(), %% regular num of batch of messages
%% num of steps
num_of_steps = ?DEF_CONSUME_NUM_OF_STEPS :: pos_integer()
}).

-record(mq_log, {
type :: atom(),
type :: atom(),
requested_at = 0 :: integer(),
format = [] :: string(),
message = [] :: string()}).
format = [] :: string(),
message = [] :: string()}).


-define(ST_IDLING, 'idling').
-define(ST_RUNNING, 'running').
-define(ST_IDLING, 'idling').
-define(ST_RUNNING, 'running').
-define(ST_SUSPENDING, 'suspending').
-type(state_of_mq() :: ?ST_IDLING |
?ST_RUNNING |
-type(state_of_mq() :: ?ST_IDLING |
?ST_RUNNING |
?ST_SUSPENDING).

-record(mq_state, {
id :: atom(),
desc = [] :: string(),
state :: [{atom(), any()}]
state :: [{atom(), any()}]
}).


-define(EVENT_RUN, 'run').
-define(EVENT_RUN, 'run').
-define(EVENT_DIAGNOSE, 'diagnose').
-define(EVENT_LOCK, 'lock').
-define(EVENT_SUSPEND, 'suspend').
-define(EVENT_RESUME, 'resume').
-define(EVENT_FINISH, 'finish').
-define(EVENT_STATE, 'state').
-define(EVENT_INCR, 'increase').
-define(EVENT_DECR, 'decrease').
-type(event_of_compaction() ::?EVENT_RUN |
-define(EVENT_LOCK, 'lock').
-define(EVENT_SUSPEND, 'suspend').
-define(EVENT_RESUME, 'resume').
-define(EVENT_FINISH, 'finish').
-define(EVENT_STATE, 'state').
-define(EVENT_INCR, 'increase').
-define(EVENT_DECR, 'decrease').
-type(event_of_compaction() ::?EVENT_RUN |
?EVENT_DIAGNOSE |
?EVENT_LOCK |
?EVENT_SUSPEND |
?EVENT_RESUME |
?EVENT_FINISH |
?EVENT_STATE |
?EVENT_INCR |
?EVENT_DECR
).

?EVENT_LOCK |
?EVENT_SUSPEND |
?EVENT_RESUME |
?EVENT_FINISH |
?EVENT_STATE |
?EVENT_INCR |
?EVENT_DECR).

-define(DEF_CHECK_MAX_INTERVAL_1, timer:seconds(1)).
-define(DEF_CHECK_MIN_INTERVAL_1, timer:seconds(0)).
Expand All @@ -152,7 +151,7 @@
end).

%% Retrieve the backend-db info
-define(DEF_DB_PATH_INDEX, "index" ).
-define(DEF_DB_PATH_INDEX, "index").
-define(DEF_DB_PATH_MESSAGE, "message").

-define(backend_db_info(Id, RootPath),
Expand Down
Loading

0 comments on commit 730a6e2

Please sign in to comment.