gproc.erl 33 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164
  1. %% ``The contents of this file are subject to the Erlang Public License,
  2. %% Version 1.1, (the "License"); you may not use this file except in
  3. %% compliance with the License. You should have received a copy of the
  4. %% Erlang Public License along with this software. If not, it can be
  5. %% retrieved via the world wide web at http://www.erlang.org/.
  6. %%
  7. %% Software distributed under the License is distributed on an "AS IS"
  8. %% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
  9. %% the License for the specific language governing rights and limitations
  10. %% under the License.
  11. %%
  12. %% The Initial Developer of the Original Code is Ericsson Utvecklings AB.
  13. %% Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
  14. %% AB. All Rights Reserved.''
  15. %%
  16. %% @author Ulf Wiger <ulf.wiger@erlang-consulting.com>
  17. %%
  18. %% @doc Extended process registry
  19. %% <p>This module implements an extended process registry</p>
  20. %% <p>For a detailed description, see
  21. %% <a href="erlang07-wiger.pdf">erlang07-wiger.pdf</a>.</p>
  22. %%
  23. %% @type type() = n | p | c | a. n = name; p = property; c = counter;
  24. %% a = aggregate_counter
  25. %% @type scope() = l | g. l = local registration; g = global registration
  26. %% @type context() = {scope(), type()} | type(). Local scope is the default
  27. %% @type sel_type() = n | p | c | a |
  28. %% names | props | counters | aggr_counters.
  29. %% @type headpat() = {keypat(),pidpat(),ValPat}.
  30. %% @type keypat() = {sel_type() | sel_var(),
  31. %% l | g | sel_var(),
  32. %% any()}.
  33. %% @type pidpat() = pid() | sel_var().
  34. %% sel_var() = DollarVar | '_'.
  35. %% @type sel_pattern() = [{headpat(), Guards, Prod}].
  36. %% @type key() = {type(), scope(), any()}
  37. %% @end
  38. -module(gproc).
  39. -behaviour(gen_server).
  40. -export([start_link/0,
  41. reg/1, reg/2, unreg/1,
  42. mreg/3,
  43. set_value/2,
  44. get_value/1,
  45. where/1,
  46. await/1, await/2,
  47. nb_wait/1,
  48. cancel_wait/2,
  49. lookup_pid/1,
  50. lookup_pids/1,
  51. lookup_values/1,
  52. update_counter/2,
  53. send/2,
  54. info/1, info/2,
  55. select/1, select/2, select/3,
  56. first/1,
  57. next/2,
  58. prev/2,
  59. last/1,
  60. table/1, table/2]).
  61. %% Convenience functions
  62. -export([add_local_name/1,
  63. add_global_name/1,
  64. add_local_property/2,
  65. add_global_property/2,
  66. add_local_counter/2,
  67. add_global_counter/2,
  68. add_local_aggr_counter/1,
  69. add_global_aggr_counter/1,
  70. lookup_local_name/1,
  71. lookup_global_name/1,
  72. lookup_local_properties/1,
  73. lookup_global_properties/1,
  74. lookup_local_counters/1,
  75. lookup_global_counters/1,
  76. lookup_local_aggr_counter/1,
  77. lookup_global_aggr_counter/1]).
  78. -export([default/1]).
  79. %%% internal exports
  80. -export([init/1,
  81. handle_cast/2,
  82. handle_call/3,
  83. handle_info/2,
  84. code_change/3,
  85. terminate/2]).
  86. -include("gproc.hrl").
  87. -define(SERVER, ?MODULE).
  88. %%-define(l, l(?LINE)). % when activated, calls a traceable empty function
  89. -define(l, ignore).
  90. -define(CHK_DIST,
  91. case whereis(gproc_dist) of
  92. undefined ->
  93. erlang:error(local_only);
  94. _ ->
  95. ok
  96. end).
  97. -record(state, {}).
  98. %% @spec () -> {ok, pid()}
  99. %%
  100. %% @doc Starts the gproc server.
  101. %%
  102. %% This function is intended to be called from gproc_sup, as part of
  103. %% starting the gproc application.
  104. %% @end
  105. start_link() ->
  106. create_tabs(),
  107. gen_server:start({local, ?SERVER}, ?MODULE, [], []).
  108. %% spec(Name::any()) -> true
  109. %%
  110. %% @doc Registers a local (unique) name. @equiv reg({n,l,Name})
  111. %% @end
  112. %%
  113. add_local_name(Name) -> reg({n,l,Name}, undefined).
  114. %% spec(Name::any()) -> true
  115. %%
  116. %% @doc Registers a global (unique) name. @equiv reg({n,g,Name})
  117. %% @end
  118. %%
  119. add_global_name(Name) -> reg({n,g,Name}, undefined).
  120. %% spec(Name::any(), Value::any()) -> true
  121. %%
  122. %% @doc Registers a local (non-unique) property. @equiv reg({p,l,Name},Value)
  123. %% @end
  124. %%
  125. add_local_property(Name , Value) -> reg({p,l,Name}, Value).
  126. %% spec(Name::any(), Value::any()) -> true
  127. %%
  128. %% @doc Registers a global (non-unique) property. @equiv reg({p,g,Name},Value)
  129. %% @end
  130. %%
  131. add_global_property(Name, Value) -> reg({p,g,Name}, Value).
  132. %% spec(Name::any(), Initial::integer()) -> true
  133. %%
  134. %% @doc Registers a local (non-unique) counter. @equiv reg({c,l,Name},Value)
  135. %% @end
  136. %%
  137. add_local_counter(Name, Initial) when is_integer(Initial) ->
  138. reg({c,l,Name}, Initial).
  139. %% spec(Name::any(), Initial::integer()) -> true
  140. %%
  141. %% @doc Registers a global (non-unique) counter. @equiv reg({c,g,Name},Value)
  142. %% @end
  143. %%
  144. add_global_counter(Name, Initial) when is_integer(Initial) ->
  145. reg({n,g,Name}, Initial).
  146. %% spec(Name::any()) -> true
  147. %%
  148. %% @doc Registers a local (unique) aggregated counter.
  149. %% @equiv reg({a,l,Name})
  150. %% @end
  151. %%
  152. add_local_aggr_counter(Name) -> reg({a,l,Name}).
  153. %% spec(Name::any()) -> true
  154. %%
  155. %% @doc Registers a global (unique) aggregated counter.
  156. %% @equiv reg({a,g,Name})
  157. %% @end
  158. %%
  159. add_global_aggr_counter(Name) -> reg({a,g,Name}).
  160. %% @spec (Name::any()) -> pid()
  161. %%
  162. %% @doc Lookup a local unique name. Fails if there is no such name.
  163. %% @equiv where({n,l,Name})
  164. %% @end
  165. %%
  166. lookup_local_name(Name) -> where({n,l,Name}).
  167. %% @spec (Name::any()) -> pid()
  168. %%
  169. %% @doc Lookup a global unique name. Fails if there is no such name.
  170. %% @equiv where({n,g,Name})
  171. %% @end
  172. %%
  173. lookup_global_name(Name) -> where({n,g,Name}).
  174. %% @spec (Name::any()) -> integer()
  175. %%
  176. %% @doc Lookup a local (unique) aggregated counter and returns its value.
  177. %% Fails if there is no such object.
  178. %% @equiv where({a,l,Name})
  179. %% @end
  180. %%
  181. lookup_local_aggr_counter(Name) -> lookup_value({a,l,Name}).
  182. %% @spec (Name::any()) -> integer()
  183. %%
  184. %% @doc Lookup a global (unique) aggregated counter and returns its value.
  185. %% Fails if there is no such object.
  186. %% @equiv where({a,g,Name})
  187. %% @end
  188. %%
  189. lookup_global_aggr_counter(Name) -> lookup_value({a,g,Name}).
  190. %% @spec (Property::any()) -> [{pid(), Value}]
  191. %%
  192. %% @doc Look up all local (non-unique) instances of a given Property.
  193. %% Returns a list of {Pid, Value} tuples for all matching objects.
  194. %% @equiv lookup_values({p, l, Property})
  195. %% @end
  196. %%
  197. lookup_local_properties(P) -> lookup_values({p,l,P}).
  198. %% @spec (Property::any()) -> [{pid(), Value}]
  199. %%
  200. %% @doc Look up all global (non-unique) instances of a given Property.
  201. %% Returns a list of {Pid, Value} tuples for all matching objects.
  202. %% @equiv lookup_values({p, g, Property})
  203. %% @end
  204. %%
  205. lookup_global_properties(P) -> lookup_values({p,g,P}).
  206. %% @spec (Counter::any()) -> [{pid(), Value::integer()}]
  207. %%
  208. %% @doc Look up all local (non-unique) instances of a given Counter.
  209. %% Returns a list of {Pid, Value} tuples for all matching objects.
  210. %% @equiv lookup_values({c, l, Counter})
  211. %% @end
  212. %%
  213. lookup_local_counters(P) -> lookup_values({c,l,P}).
  214. %% @spec (Counter::any()) -> [{pid(), Value::integer()}]
  215. %%
  216. %% @doc Look up all global (non-unique) instances of a given Counter.
  217. %% Returns a list of {Pid, Value} tuples for all matching objects.
  218. %% @equiv lookup_values({c, g, Counter})
  219. %% @end
  220. %%
  221. lookup_global_counters(P) -> lookup_values({c,g,P}).
  222. %% @spec reg(Key::key()) -> true
  223. %%
  224. %% @doc
  225. %% @equiv reg(Key, default(Key))
  226. %% @end
  227. reg(Key) ->
  228. reg(Key, default(Key)).
  229. default({T,_,_}) when T==c -> 0;
  230. default(_) -> undefined.
  231. %% @spec await(Key::key(), Timeout) -> {pid(),Value}
  232. %% Timeout = integer() | infinity
  233. %%
  234. %% @doc Wait for a local name to be registered.
  235. %% The function raises an exception if the timeout expires. Timeout must be
  236. %% either an interger &gt; 0 or 'infinity'.
  237. %% A small optimization: we first perform a lookup, to see if the name
  238. %% is already registered. This way, the cost of the operation will be
  239. %% roughly the same as of where/1 in the case where the name is already
  240. %% registered (the difference: await/2 also returns the value).
  241. %% @end
  242. %%
  243. await(Key) ->
  244. await(Key, infinity).
  245. await({n,l,_} = Key, Timeout) ->
  246. case ets:lookup(?TAB, {Key, n}) of
  247. [{_, Pid, Value}] ->
  248. {Pid, Value};
  249. _ ->
  250. request_wait(Key, Timeout)
  251. end;
  252. await(K, T) ->
  253. erlang:error(badarg, [K, T]).
  254. request_wait({n,l,_} = Key, Timeout) ->
  255. TRef = case Timeout of
  256. infinity -> no_timer;
  257. T when is_integer(T), T > 0 ->
  258. erlang:start_timer(T, self(), timeout);
  259. _ ->
  260. erlang:error(badarg, [Key, Timeout])
  261. end,
  262. WRef = call({await,Key}),
  263. receive
  264. {gproc, WRef, registered, {_K, Pid, V}} ->
  265. {Pid, V};
  266. {timeout, TRef, timeout} ->
  267. cancel_wait(Key, WRef),
  268. erlang:error(timeout, [Key, Timeout])
  269. end.
  270. %% @spec nb_wait(Key::key()) -> Ref
  271. %%
  272. %% @doc Wait for a local name to be registered.
  273. %% The caller can expect to receive a message,
  274. %% {gproc, Ref, registered, {Key, Pid, Value}}, once the name is registered.
  275. %% @end
  276. %%
  277. nb_wait({n,l,_} = Key) ->
  278. call({await, Key});
  279. nb_wait(Key) ->
  280. erlang:error(badarg, [Key]).
  281. cancel_wait(Key, Ref) ->
  282. cast({cancel_wait, self(), Key, Ref}),
  283. ok.
  284. %% @spec reg(Key::key(), Value) -> true
  285. %%
  286. %% @doc Register a name or property for the current process
  287. %%
  288. %%
  289. reg({_,g,_} = Key, Value) ->
  290. %% anything global
  291. ?CHK_DIST,
  292. gproc_dist:reg(Key, Value);
  293. reg({p,l,_} = Key, Value) ->
  294. local_reg(Key, Value);
  295. reg({a,l,_} = Key, undefined) ->
  296. call({reg, Key, undefined});
  297. reg({c,l,_} = Key, Value) when is_integer(Value) ->
  298. call({reg, Key, Value});
  299. reg({n,l,_} = Key, Value) ->
  300. call({reg, Key, Value});
  301. reg(_, _) ->
  302. erlang:error(badarg).
  303. %% @spec mreg(type(), scope(), [{Key::any(), Value::any()}]) -> true
  304. %%
  305. %% @doc Register multiple {Key,Value} pairs of a given type and scope.
  306. %%
  307. %% This function is more efficient than calling {@link reg/2} repeatedly.
  308. %% @end
  309. mreg(T, g, KVL) ->
  310. ?CHK_DIST,
  311. gproc_dist:mreg(T, KVL);
  312. mreg(T, l, KVL) when T==a; T==n ->
  313. if is_list(KVL) ->
  314. call({mreg, T, l, KVL});
  315. true ->
  316. erlang:error(badarg)
  317. end;
  318. mreg(p, l, KVL) ->
  319. local_mreg(p, KVL);
  320. mreg(_, _, _) ->
  321. erlang:error(badarg).
  322. %% @spec (Key:: key()) -> true
  323. %%
  324. %% @doc Unregister a name or property.
  325. %% @end
  326. unreg(Key) ->
  327. case Key of
  328. {_, g, _} ->
  329. ?CHK_DIST,
  330. gproc_dist:unreg(Key);
  331. {T, l, _} when T == n;
  332. T == a -> call({unreg, Key});
  333. {_, l, _} ->
  334. case ets:member(?TAB, {Key,self()}) of
  335. true ->
  336. gproc_lib:remove_reg(Key, self());
  337. false ->
  338. erlang:error(badarg)
  339. end
  340. end.
  341. %% @spec (select_pattern()) -> list(sel_object())
  342. %% @doc
  343. %% @equiv select(all, Pat)
  344. %% @end
  345. select(Pat) ->
  346. select(all, Pat).
  347. %% @spec (Type::sel_type(), Pat::sel_pattern()) -> [{Key, Pid, Value}]
  348. %%
  349. %% @doc Perform a select operation on the process registry.
  350. %%
  351. %% The physical representation in the registry may differ from the above,
  352. %% but the select patterns are transformed appropriately.
  353. %% @end
  354. select(Type, Pat) ->
  355. ets:select(?TAB, pattern(Pat, Type)).
  356. %% @spec (Type::sel_type(), Pat::sel_patten(), Limit::integer()) ->
  357. %% [{Key, Pid, Value}]
  358. %% @doc Like {@link select/2} but returns Limit objects at a time.
  359. %%
  360. %% See [http://www.erlang.org/doc/man/ets.html#select-3].
  361. %% @end
  362. select(Type, Pat, Limit) ->
  363. ets:select(?TAB, pattern(Pat, Type), Limit).
  364. %%% Local properties can be registered in the local process, since
  365. %%% no other process can interfere.
  366. %%%
  367. local_reg(Key, Value) ->
  368. case gproc_lib:insert_reg(Key, Value, self(), l) of
  369. false -> erlang:error(badarg);
  370. true -> monitor_me()
  371. end.
  372. local_mreg(_, []) -> true;
  373. local_mreg(T, [_|_] = KVL) ->
  374. case gproc_lib:insert_many(T, l, KVL, self()) of
  375. false -> erlang:error(badarg);
  376. {true,_} -> monitor_me()
  377. end.
  378. %% @spec (Key :: key(), Value) -> true
  379. %% @doc Sets the value of the registeration entry given by Key
  380. %%
  381. %% Key is assumed to exist and belong to the calling process.
  382. %% If it doesn't, this function will exit.
  383. %%
  384. %% Value can be any term, unless the object is a counter, in which case
  385. %% it must be an integer.
  386. %% @end
  387. %%
  388. set_value({_,g,_} = Key, Value) ->
  389. ?CHK_DIST,
  390. gproc_dist:set_value(Key, Value);
  391. set_value({a,l,_} = Key, Value) when is_integer(Value) ->
  392. call({set, Key, Value});
  393. set_value({n,l,_} = Key, Value) ->
  394. %% we cannot do this locally, since we have to check that the object
  395. %% exists first - not an atomic update.
  396. call({set, Key, Value});
  397. set_value({p,l,_} = Key, Value) ->
  398. %% we _can_ to this locally, since there is no race condition - no
  399. %% other process can update our properties.
  400. case gproc_lib:do_set_value(Key, Value, self()) of
  401. true -> true;
  402. false ->
  403. erlang:error(badarg)
  404. end;
  405. set_value({c,l,_} = Key, Value) when is_integer(Value) ->
  406. gproc_lib:do_set_counter_value(Key, Value, self());
  407. set_value(_, _) ->
  408. erlang:error(badarg).
  409. %% @spec (Key) -> Value
  410. %% @doc Read the value stored with a key registered to the current process.
  411. %%
  412. %% If no such key is registered to the current process, this function exits.
  413. %% @end
  414. get_value(Key) ->
  415. get_value(Key, self()).
  416. get_value({T,_,_} = Key, Pid) when is_pid(Pid) ->
  417. if T==n orelse T==a ->
  418. case ets:lookup(?TAB, {Key, T}) of
  419. [{_, P, Value}] when P == Pid -> Value;
  420. _ -> erlang:error(badarg)
  421. end;
  422. true ->
  423. ets:lookup_element(?TAB, {Key, Pid}, 3)
  424. end;
  425. get_value(_, _) ->
  426. erlang:error(badarg).
  427. %% @spec (Key) -> Pid
  428. %% @doc Lookup the Pid stored with a key.
  429. %%
  430. lookup_pid({_T,_,_} = Key) ->
  431. case where(Key) of
  432. undefined -> erlang:error(badarg);
  433. P -> P
  434. end.
  435. lookup_value({T,_,_} = Key) ->
  436. if T==n orelse T==a ->
  437. ets:lookup_element(?TAB, {Key,T}, 3);
  438. true ->
  439. erlang:error(badarg)
  440. end.
  441. %% @spec (Key::key()) -> pid()
  442. %%
  443. %% @doc Returns the pid registered as Key
  444. %%
  445. %% The type of registration entry must be either name or aggregated counter.
  446. %% Otherwise this function will exit. Use {@link lookup_pids/1} in these
  447. %% cases.
  448. %% @end
  449. %%
  450. where({T,_,_}=Key) ->
  451. if T==n orelse T==a ->
  452. case ets:lookup(?TAB, {Key,T}) of
  453. [{_, P, _Value}] ->
  454. P;
  455. _ -> % may be [] or [{Key,Waiters}]
  456. undefined
  457. end;
  458. true ->
  459. erlang:error(badarg)
  460. end.
  461. %% @spec (Key::key()) -> [pid()]
  462. %%
  463. %% @doc Returns a list of pids with the published key Key
  464. %%
  465. %% If the type of registration entry is either name or aggregated counter,
  466. %% this function will return either an empty list, or a list of one pid.
  467. %% For non-unique types, the return value can be a list of any length.
  468. %% @end
  469. %%
  470. lookup_pids({T,_,_} = Key) ->
  471. if T==n orelse T==a ->
  472. ets:select(?TAB, [{{{Key,T}, '$1', '_'},[],['$1']}]);
  473. true ->
  474. ets:select(?TAB, [{{{Key,'_'}, '$1', '_'},[],['$1']}])
  475. %%% true ->
  476. %%% erlang:error(badarg)
  477. end.
  478. %% @spec (Key::key()) -> [{pid(), Value}]
  479. %%
  480. %% @doc Retrieve the `{Pid,Value}' pairs corresponding to Key.
  481. %%
  482. %% Key refer to any type of registry object. If it refers to a unique
  483. %% object, the list will be of length 0 or 1. If it refers to a non-unique
  484. %% object, the return value can be a list of any length.
  485. %% @end
  486. %%
  487. lookup_values({T,_,_} = Key) ->
  488. if T==n orelse T==a ->
  489. ets:select(?TAB, [{{{Key,T}, '$1', '$2'},[],[{{'$1','$2'}}]}]);
  490. true ->
  491. ets:select(?TAB, [{{{Key,'_'}, '$1', '$2'},[],[{{'$1','$2'}}]}])
  492. end.
  493. %% @spec (Key::key(), Incr::integer()) -> integer()
  494. %%
  495. %% @doc Updates the counter registered as Key for the current process.
  496. %%
  497. %% This function works like ets:update_counter/3
  498. %% (see [http://www.erlang.org/doc/man/ets.html#update_counter-3]), but
  499. %% will fail if the type of object referred to by Key is not a counter.
  500. %% @end
  501. %%
  502. update_counter({c,l,_} = Key, Incr) when is_integer(Incr) ->
  503. gproc_lib:update_counter(Key, Incr, self());
  504. update_counter({c,g,_} = Key, Incr) when is_integer(Incr) ->
  505. ?CHK_DIST,
  506. gproc_dist:update_counter(Key, Incr);
  507. update_counter(_, _) ->
  508. erlang:error(badarg).
  509. %% @spec (Key::key(), Msg::any()) -> Msg
  510. %%
  511. %% @doc Sends a message to the process, or processes, corresponding to Key.
  512. %%
  513. %% If Key belongs to a unique object (name or aggregated counter), this
  514. %% function will send a message to the corresponding process, or fail if there
  515. %% is no such process. If Key is for a non-unique object type (counter or
  516. %% property), Msg will be send to all processes that have such an object.
  517. %% @end
  518. %%
  519. send({T,C,_} = Key, Msg) when C==l; C==g ->
  520. if T == n orelse T == a ->
  521. case ets:lookup(?TAB, {Key, T}) of
  522. [{_, Pid, _}] ->
  523. Pid ! Msg;
  524. _ ->
  525. erlang:error(badarg)
  526. end;
  527. T==p orelse T==c ->
  528. %% BUG - if the key part contains select wildcards, we may end up
  529. %% sending multiple messages to the same pid
  530. lists:foreach(fun(Pid) ->
  531. Pid ! Msg
  532. end, lookup_pids(Key)),
  533. Msg;
  534. true ->
  535. erlang:error(badarg)
  536. end;
  537. send(_, _) ->
  538. erlang:error(badarg).
  539. %% @spec (Type :: type()) -> key() | '$end_of_table'
  540. %%
  541. %% @doc Behaves as ets:first(Tab) for a given type of registration object.
  542. %%
  543. %% See [http://www.erlang.org/doc/man/ets.html#first-1].
  544. %% The registry behaves as an ordered_set table.
  545. %% @end
  546. %%
  547. first(Type) ->
  548. {HeadPat,_} = headpat(Type, '_', '_', '_'),
  549. case ets:select(?TAB, [{HeadPat,[],[{element,1,'$_'}]}], 1) of
  550. {[First], _} ->
  551. First;
  552. _ ->
  553. '$end_of_table'
  554. end.
  555. %% @spec (Context :: context()) -> key() | '$end_of_table'
  556. %%
  557. %% @doc Behaves as ets:last(Tab) for a given type of registration object.
  558. %%
  559. %% See [http://www.erlang.org/doc/man/ets.html#last-1].
  560. %% The registry behaves as an ordered_set table.
  561. %% @end
  562. %%
  563. last(Context) ->
  564. {S, T} = get_s_t(Context),
  565. S1 = if S == '_'; S == l -> m;
  566. S == g -> h
  567. end,
  568. Beyond = {{T,S1,[]},[]},
  569. step(ets:prev(?TAB, Beyond), S, T).
  570. %% @spec (Context::context(), Key::key()) -> key() | '$end_of_table'
  571. %%
  572. %% @doc Behaves as ets:next(Tab,Key) for a given type of registration object.
  573. %%
  574. %% See [http://www.erlang.org/doc/man/ets.html#next-2].
  575. %% The registry behaves as an ordered_set table.
  576. %% @end
  577. %%
  578. next(Context, K) ->
  579. {S,T} = get_s_t(Context),
  580. step(ets:next(?TAB,K), S, T).
  581. %% @spec (Context::context(), Key::key()) -> key() | '$end_of_table'
  582. %%
  583. %% @doc Behaves as ets:prev(Tab,Key) for a given type of registration object.
  584. %%
  585. %% See [http://www.erlang.org/doc/man/ets.html#prev-2].
  586. %% The registry behaves as an ordered_set table.
  587. %% @end
  588. %%
  589. prev(Context, K) ->
  590. {S, T} = get_s_t(Context),
  591. step(ets:prev(?TAB, K), S, T).
  592. step(Key, '_', '_') ->
  593. case Key of
  594. {{_,_,_},_} -> Key;
  595. _ -> '$end_of_table'
  596. end;
  597. step(Key, '_', T) ->
  598. case Key of
  599. {{T,_,_},_} -> Key;
  600. _ -> '$end_of_table'
  601. end;
  602. step(Key, S, '_') ->
  603. case Key of
  604. {{_, S, _}, _} -> Key;
  605. _ -> '$end_of_table'
  606. end;
  607. step(Key, S, T) ->
  608. case Key of
  609. {{T, S, _}, _} -> Key;
  610. _ -> '$end_of_table'
  611. end.
  612. %% @spec (Pid::pid()) -> ProcessInfo
  613. %% ProcessInfo = [{gproc, [{Key,Value}]} | ProcessInfo]
  614. %%
  615. %% @doc Similar to `process_info(Pid)' but with additional gproc info.
  616. %%
  617. %% Returns the same information as process_info(Pid), but with the
  618. %% addition of a `gproc' information item, containing the `{Key,Value}'
  619. %% pairs registered to the process.
  620. %% @end
  621. info(Pid) when is_pid(Pid) ->
  622. Items = [?MODULE | [ I || {I,_} <- process_info(self())]],
  623. [info(Pid,I) || I <- Items].
  624. %% @spec (Pid::pid(), Item::atom()) -> {Item, Info}
  625. %%
  626. %% @doc Similar to process_info(Pid, Item), but with additional gproc info.
  627. %%
  628. %% For `Item = gproc', this function returns a list of `{Key, Value}' pairs
  629. %% registered to the process Pid. For other values of Item, it returns the
  630. %% same as [http://www.erlang.org/doc/man/erlang.html#process_info-2].
  631. %% @end
  632. info(Pid, ?MODULE) ->
  633. Keys = ets:select(?TAB, [{ {{Pid,'$1'}}, [], ['$1'] }]),
  634. {?MODULE, lists:zf(
  635. fun(K) ->
  636. try V = get_value(K, Pid),
  637. {true, {K,V}}
  638. catch
  639. error:_ ->
  640. false
  641. end
  642. end, Keys)};
  643. info(Pid, I) ->
  644. process_info(Pid, I).
  645. %%% ==========================================================
  646. %% @hidden
  647. handle_cast({monitor_me, Pid}, S) ->
  648. erlang:monitor(process, Pid),
  649. {noreply, S};
  650. handle_cast({cancel_wait, Pid, {T,_,_} = Key, Ref}, S) ->
  651. case ets:lookup(?TAB, {Key,T}) of
  652. [{K, Waiters}] ->
  653. NewWaiters = Waiters -- [{Pid,Ref}],
  654. %% for now, we don't remove the reverse entry. If we should do
  655. %% that, we have to make sure that Pid doesn't have another
  656. %% waiter (which it shouldn't have, given that the wait is
  657. %% synchronous). Keeping it is not problematic - worst case, we
  658. %% will get an unnecessary cleanup.
  659. ets:insert(?TAB, {K, NewWaiters});
  660. _ ->
  661. ignore
  662. end,
  663. {noreply, S}.
  664. %% @hidden
  665. handle_call({reg, {_T,l,_} = Key, Val}, {Pid,_}, S) ->
  666. case try_insert_reg(Key, Val, Pid) of
  667. true ->
  668. ensure_monitor(Pid),
  669. {reply, true, S};
  670. false ->
  671. {reply, badarg, S}
  672. end;
  673. handle_call({unreg, {_,l,_} = Key}, {Pid,_}, S) ->
  674. case ets:member(?TAB, {Pid,Key}) of
  675. true ->
  676. gproc_lib:remove_reg(Key, Pid),
  677. {reply, true, S};
  678. false ->
  679. {reply, badarg, S}
  680. end;
  681. handle_call({await, {T,l,_} = Key}, {Pid,Ref} = From, S) ->
  682. Rev = {{Pid,Key}, r},
  683. case ets:lookup(?TAB, {Key,T}) of
  684. [{_, P, Value}] ->
  685. %% for symmetry, we always reply with Ref and then send a message
  686. gen_server:reply(From, Ref),
  687. Pid ! {gproc, Ref, registered, {Key, P, Value}},
  688. {noreply, S};
  689. [{K, Waiters}] ->
  690. NewWaiters = [{Pid,Ref} | Waiters],
  691. ets:insert(?TAB, [{K, NewWaiters}, Rev]),
  692. gproc_lib:ensure_monitor(Pid),
  693. {reply, Ref, S};
  694. [] ->
  695. ets:insert(?TAB, [{{Key,T}, [{Pid,Ref}]}, Rev]),
  696. gproc_lib:ensure_monitor(Pid),
  697. {reply, Ref, S}
  698. end;
  699. handle_call({mreg, T, l, L}, {Pid,_}, S) ->
  700. try gproc_lib:insert_many(T, l, L, Pid) of
  701. {true,_} -> {reply, true, S};
  702. false -> {reply, badarg, S}
  703. catch
  704. error:_ -> {reply, badarg, S}
  705. end;
  706. handle_call({set, {_,l,_} = Key, Value}, {Pid,_}, S) ->
  707. case gproc_lib:do_set_value(Key, Value, Pid) of
  708. true ->
  709. {reply, true, S};
  710. false ->
  711. {reply, badarg, S}
  712. end;
  713. handle_call(_, _, S) ->
  714. {reply, badarg, S}.
  715. %% @hidden
  716. handle_info({'DOWN', _MRef, process, Pid, _}, S) ->
  717. process_is_down(Pid),
  718. {noreply, S};
  719. handle_info(_, S) ->
  720. {noreply, S}.
  721. %% @hidden
  722. code_change(_FromVsn, S, _Extra) ->
  723. {ok, S}.
  724. %% @hidden
  725. terminate(_Reason, _S) ->
  726. ok.
  727. call(Req) ->
  728. case gen_server:call(?MODULE, Req) of
  729. badarg -> erlang:error(badarg, Req);
  730. Reply -> Reply
  731. end.
  732. cast(Msg) ->
  733. gen_server:cast(?MODULE, Msg).
  734. try_insert_reg({T,l,_} = Key, Val, Pid) ->
  735. case gproc_lib:insert_reg(Key, Val, Pid, l) of
  736. false ->
  737. case ets:lookup(?TAB, {Key,T}) of
  738. %% In this particular case, the lookup cannot result in
  739. %% [{_, Waiters}], since the insert_reg/4 function would
  740. %% have succeeded then.
  741. [{_, OtherPid, _}] ->
  742. case is_process_alive(OtherPid) of
  743. true ->
  744. false;
  745. false ->
  746. process_is_down(Pid),
  747. true = gproc_lib:insert_reg(Key, Val, Pid, l)
  748. end;
  749. [] ->
  750. false
  751. end;
  752. true ->
  753. true
  754. end.
  755. process_is_down(Pid) ->
  756. Keys = ets:select(?TAB, [{{{Pid,'$1'},'$2'},
  757. [{'==',{element,2,'$1'},l}], [{{'$1','$2'}}]}]),
  758. ets:select_delete(?TAB, [{{{Pid,{'_',l,'_'}},'_'}, [], [true]}]),
  759. ets:delete(?TAB, Pid),
  760. lists:foreach(fun({Key,r}) ->
  761. gproc_lib:remove_reg_1(Key, Pid);
  762. ({Key,w}) ->
  763. gproc_lib:remove_waiter(Key, Pid)
  764. end, Keys).
  765. create_tabs() ->
  766. ets:new(?MODULE, [ordered_set, public, named_table]).
  767. %% @hidden
  768. init([]) ->
  769. {ok, #state{}}.
  770. ensure_monitor(Pid) when node(Pid) == node() ->
  771. case ets:insert_new(?TAB, {Pid}) of
  772. false -> ok;
  773. true -> erlang:monitor(process, Pid)
  774. end;
  775. ensure_monitor(_) ->
  776. true.
  777. monitor_me() ->
  778. case ets:insert_new(?TAB, {self()}) of
  779. false -> true;
  780. true ->
  781. cast({monitor_me,self()}),
  782. true
  783. end.
  784. pattern([{'_', Gs, As}], T) ->
  785. ?l,
  786. {HeadPat, Vs} = headpat(T, '$1', '$2', '$3'),
  787. [{HeadPat, rewrite(Gs,Vs), rewrite(As,Vs)}];
  788. pattern([{{A,B,C},Gs,As}], Scope) ->
  789. ?l,
  790. {HeadPat, Vars} = headpat(Scope, A,B,C),
  791. [{HeadPat, rewrite(Gs,Vars), rewrite(As,Vars)}];
  792. pattern([{Head, Gs, As}], Scope) ->
  793. ?l,
  794. case is_var(Head) of
  795. {true,_N} ->
  796. HeadPat = {{{type(Scope),'_','_'},'_'},'_','_'},
  797. Vs = [{Head, obj_prod()}],
  798. %% {HeadPat, Vs} = headpat(Scope, A,B,C),
  799. %% the headpat function should somehow verify that Head is
  800. %% consistent with Scope (or should we add a guard?)
  801. [{HeadPat, rewrite(Gs, Vs), rewrite(As, Vs)}];
  802. false ->
  803. erlang:error(badarg)
  804. end.
  805. %% This is the expression to use in guards and the RHS to address the whole
  806. %% object, in its logical representation.
  807. obj_prod() ->
  808. {{ {element,1,{element,1,'$_'}},
  809. {element,2,'$_'},
  810. {element,3,'$_'} }}.
  811. obj_prod_l() ->
  812. [ {element,1,{element,1,'$_'}},
  813. {element,2,'$_'},
  814. {element,3,'$_'} ].
  815. headpat({S, T}, V1,V2,V3) when S==global; S==local; S==all ->
  816. headpat(type(T), scope(S), V1,V2,V3);
  817. headpat(T, V1, V2, V3) when is_atom(T) ->
  818. headpat(type(T), l, V1, V2, V3);
  819. headpat(_, _, _, _) -> erlang:error(badarg).
  820. headpat(T, C, V1,V2,V3) ->
  821. Rf = fun(Pos) ->
  822. {element,Pos,{element,1,{element,1,'$_'}}}
  823. end,
  824. K2 = if T==n orelse T==a -> T;
  825. true -> '_'
  826. end,
  827. {Kp,Vars} = case V1 of
  828. {Vt,Vc,Vn} ->
  829. ?l,
  830. {T1,Vs1} = subst(T,Vt,fun() -> Rf(1) end,[]),
  831. {C1,Vs2} = subst(C,Vc,fun() -> Rf(2) end,Vs1),
  832. {{T1,C1,Vn}, Vs2};
  833. '_' ->
  834. ?l,
  835. {{T,C,'_'}, []};
  836. _ ->
  837. ?l,
  838. case is_var(V1) of
  839. {true,_} ->
  840. {{T,C,V1}, [{V1, {element,1,
  841. {element,1,'$_'}}}]};
  842. false ->
  843. erlang:error(badarg)
  844. end
  845. end,
  846. {{{Kp,K2},V2,V3}, Vars}.
  847. %% l(L) -> L.
  848. subst(X, '_', _F, Vs) ->
  849. {X, Vs};
  850. subst(X, V, F, Vs) ->
  851. case is_var(V) of
  852. {true,_} ->
  853. {X, [{V,F()}|Vs]};
  854. false ->
  855. {V, Vs}
  856. end.
  857. scope(all) -> '_';
  858. scope(global) -> g;
  859. scope(local) -> l.
  860. type(all) -> '_';
  861. type(T) when T==n; T==p; T==c; T==a -> T;
  862. type(names) -> n;
  863. type(props) -> p;
  864. type(counters) -> c;
  865. type(aggr_counters) -> a.
  866. keypat(Context) ->
  867. {S,T} = get_s_t(Context),
  868. {{T,S,'_'},'_'}.
  869. get_s_t({S,T}) -> {scope(S), type(T)};
  870. get_s_t(T) when is_atom(T) ->
  871. {l, type(T)}.
  872. is_var('$1') -> {true,1};
  873. is_var('$2') -> {true,2};
  874. is_var('$3') -> {true,3};
  875. is_var('$4') -> {true,4};
  876. is_var('$5') -> {true,5};
  877. is_var('$6') -> {true,6};
  878. is_var('$7') -> {true,7};
  879. is_var('$8') -> {true,8};
  880. is_var('$9') -> {true,9};
  881. is_var(X) when is_atom(X) ->
  882. case atom_to_list(X) of
  883. "$" ++ Tl ->
  884. try N = list_to_integer(Tl),
  885. {true,N}
  886. catch
  887. error:_ ->
  888. false
  889. end;
  890. _ ->
  891. false
  892. end;
  893. is_var(_) -> false.
  894. rewrite(Gs, R) ->
  895. [rewrite1(G, R) || G <- Gs].
  896. rewrite1('$_', _) ->
  897. obj_prod();
  898. rewrite1('$$', _) ->
  899. obj_prod_l();
  900. rewrite1(Guard, R) when is_tuple(Guard) ->
  901. list_to_tuple([rewrite1(G, R) || G <- tuple_to_list(Guard)]);
  902. rewrite1(Exprs, R) when is_list(Exprs) ->
  903. [rewrite1(E, R) || E <- Exprs];
  904. rewrite1(V, R) when is_atom(V) ->
  905. case is_var(V) of
  906. {true,_N} ->
  907. case lists:keysearch(V, 1, R) of
  908. {value, {_, V1}} ->
  909. V1;
  910. false ->
  911. V
  912. end;
  913. false ->
  914. V
  915. end;
  916. rewrite1(Expr, _) ->
  917. Expr.
  918. %% @spec (Context::context()) -> any()
  919. %%
  920. %% @doc
  921. %% @equiv table(Context, [])
  922. %% @end
  923. %%
  924. table(Context) ->
  925. table(Context, []).
  926. %% @spec (Context::context(), Opts) -> any()
  927. %%
  928. %% @doc QLC table generator for the gproc registry.
  929. %% Context specifies which subset of the registry should be queried.
  930. %% See [http://www.erlang.org/doc/man/qlc.html].
  931. %% @end
  932. table(Ctxt, Opts) ->
  933. [Traverse, NObjs] = [proplists:get_value(K,Opts,Def) ||
  934. {K,Def} <- [{traverse,select}, {n_objects,100}]],
  935. TF = case Traverse of
  936. first_next ->
  937. fun() -> qlc_next(Ctxt, first(Ctxt)) end;
  938. last_prev -> fun() -> qlc_prev(Ctxt, last(Ctxt)) end;
  939. select ->
  940. fun(MS) -> qlc_select(select(Ctxt, MS, NObjs)) end;
  941. {select,MS} ->
  942. fun() -> qlc_select(select(Ctxt, MS, NObjs)) end;
  943. _ ->
  944. erlang:error(badarg, [Ctxt,Opts])
  945. end,
  946. InfoFun = fun(indices) -> [2];
  947. (is_unique_objects) -> is_unique(Ctxt);
  948. (keypos) -> 1;
  949. (is_sorted_key) -> true;
  950. (num_of_objects) ->
  951. %% this is just a guesstimate.
  952. trunc(ets:info(?TAB,size) / 2.5)
  953. end,
  954. LookupFun =
  955. case Traverse of
  956. {select, _MS} -> undefined;
  957. _ -> fun(Pos, Ks) -> qlc_lookup(Ctxt, Pos, Ks) end
  958. end,
  959. qlc:table(TF, [{info_fun, InfoFun},
  960. {lookup_fun, LookupFun}] ++ [{K,V} || {K,V} <- Opts,
  961. K =/= traverse,
  962. K =/= n_objects]).
  963. qlc_lookup(_Scope, 1, Keys) ->
  964. lists:flatmap(
  965. fun(Key) ->
  966. ets:select(?TAB, [{ {{Key,'_'},'_','_'}, [],
  967. [{{ {element,1,{element,1,'$_'}},
  968. {element,2,'$_'},
  969. {element,3,'$_'} }}] }])
  970. end, Keys);
  971. qlc_lookup(Scope, 2, Pids) ->
  972. lists:flatmap(fun(Pid) ->
  973. Found =
  974. ets:select(?TAB, [{ {{Pid,keypat(Scope)}},
  975. [], ['$_']}]),
  976. lists:flatmap(
  977. fun({{_,{T,_,_}=K}}) ->
  978. K2 = if T==n orelse T==a -> T;
  979. true -> Pid
  980. end,
  981. case ets:lookup(?TAB, {K,K2}) of
  982. [{{Key,_},_,Value}] ->
  983. [{Key, Pid, Value}];
  984. [] ->
  985. []
  986. end
  987. end, Found)
  988. end, Pids).
  989. qlc_next(_, '$end_of_table') -> [];
  990. qlc_next(Scope, K) ->
  991. case ets:lookup(?TAB, K) of
  992. [{{Key,_}, Pid, V}] ->
  993. [{Key,Pid,V} | fun() -> qlc_next(Scope, next(Scope, K)) end];
  994. [] ->
  995. qlc_next(Scope, next(Scope, K))
  996. end.
  997. qlc_prev(_, '$end_of_table') -> [];
  998. qlc_prev(Scope, K) ->
  999. case ets:lookup(?TAB, K) of
  1000. [{{Key,_},Pid,V}] ->
  1001. [{Key,Pid,V} | fun() -> qlc_prev(Scope, prev(Scope, K)) end];
  1002. [] ->
  1003. qlc_prev(Scope, prev(Scope, K))
  1004. end.
  1005. qlc_select('$end_of_table') ->
  1006. [];
  1007. qlc_select({Objects, Cont}) ->
  1008. Objects ++ fun() -> qlc_select(ets:select(Cont)) end.
  1009. is_unique(names) -> true;
  1010. is_unique(aggr_counters) -> true;
  1011. is_unique({_, names}) -> true;
  1012. is_unique({_, aggr_counters}) -> true;
  1013. is_unique(n) -> true;
  1014. is_unique(a) -> true;
  1015. is_unique({_,n}) -> true;
  1016. is_unique({_,a}) -> true;
  1017. is_unique(_) -> false.