🤞🏿 ❌ ⛅️ धूमकेतु - 1,000,000 उपयोगकर्ताओं के भार के साथ मोचीवेब के लिए आवेदन। भाग 3/3 😊 🙇🏼 🏬

लेखों की इस श्रृंखला में भाग 1 और भाग 2 में दिखाया गया है कि मोचीवेब का उपयोग करके एप्लिकेशन कैसे बनाया जाए, और कनेक्टेड उपयोगकर्ताओं को संदेश कैसे भेजें। हमने प्रत्येक कनेक्शन के लिए मेमोरी की खपत को 8 KB तक घटा दिया। हमने c10k परीक्षण दोहराया। हमने शेड्यूल बनाया। यह मजेदार था, लेकिन अब 1 मिलियन कनेक्शन के लिए सब कुछ दोहराने का समय है।

इस लेख में निम्नलिखित शामिल हैं:

• मानेसिया डेटाबेस का उपयोग;

• एक मिलियन उपयोगकर्ताओं के लिए फ़ॉर्म "मित्रों" के एक प्रशंसनीय डेटा सेट की पीढ़ी;

• Mnesia को कॉन्फ़िगर करें और हमारे डेटा को दर्ज करें;

• एक मशीन से एक लाख कनेक्शनों की खोज;

• 1 मिलियन उपयोगकर्ताओं के साथ तुलनात्मक परीक्षण;

• प्रसंस्करण कनेक्शन के लिए लिबवेंट + सी;

• अंतिम निष्कर्ष।

इस परीक्षण का एक हिस्सा एकल परीक्षण मशीन से 1,000,000 कनेक्शन खोलने की क्षमता है। एक सर्वर लिखना जो 1,000,000 कनेक्शन को स्वीकार करने में सक्षम है, वास्तव में 1,000,000 कनेक्शन बनाना आसान है। इस प्रकार, इस लेख की एक उचित राशि एकल मशीन से 1,000,000 कनेक्शन खोलने के लिए उपयोग किए जाने वाले तरीकों के बारे में है।

हमारे Pubsub का शुभारंभ।

भाग 2 में, हमने विशिष्ट उपयोगकर्ताओं को संदेश भेजने के लिए एक राउटर का उपयोग किया। यह चैट / आईएम प्रणाली के लिए बहुत अच्छा है, लेकिन इसके बजाय और भी उपयोगी चीजें हैं जो हम कर सकते हैं। इससे पहले कि हम बड़े पैमाने पर परीक्षण शुरू करें, चलो एक और मॉड्यूल जोड़ते हैं - ग्राहक डेटाबेस। हम आपके दोस्तों के बारे में डेटा के साथ एक रिपॉजिटरी बनाएंगे, इसलिए यह आपको आपके दोस्तों की सूची के लोगों द्वारा उत्पन्न सभी घटनाओं के साथ प्रदान कर सकता है।

मेरा इरादा Last.fm के लिए इसका उपयोग करना है। इस तरह, मैं उन गीतों का एक वास्तविक समय चैनल प्राप्त कर सकता हूं जो मेरे मित्र वर्तमान में सुन रहे हैं। यह सामाजिक नेटवर्क पर उत्पन्न अन्य घटनाओं पर समान रूप से लागू हो सकता है। फ़्लिकर, फेसबुक, ट्विटर, आदि के तत्वों को अपलोड की गई तस्वीरें समाचार फ़ीड FriendFeed में एक वास्तविक समय बीटा API भी है, इसलिए यह निश्चित रूप से प्रासंगिक है।

एक सदस्यता प्रबंधक को लागू करना

हम एक साधारण सदस्यता प्रबंधक को लागू करते हैं, लेकिन हम स्वचालित रूप से अपने सभी दोस्तों को लोगों की सदस्यता लेंगे।

एपीआई:

• add_subsults ([{सब्सक्राइबर, सब्सक्राइबर}, ...])

• remove_subsults ([{सब्सक्राइबर, सब्सक्राइबर}, ...])

• get_subscribers (उपयोगकर्ता)

-module(subsmanager). -behaviour(gen_server). -include("/usr/local/lib/erlang/lib/stdlib-1.15.4/include/qlc.hrl"). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -export([add_subscriptions/1, remove_subscriptions/1, get_subscribers/1, first_run/0, stop/0, start_link/0]). -record(subscription, {subscriber, subscribee}). -record(state, {}). % state is all in mnesia -define(SERVER, global:whereis_name(?MODULE)). start_link() -> gen_server:start_link({global, ?MODULE}, ?MODULE, [], []). stop() -> gen_server:call(?SERVER, {stop}). add_subscriptions(SubsList) -> gen_server:call(?SERVER, {add_subscriptions, SubsList}, infinity). remove_subscriptions(SubsList) -> gen_server:call(?SERVER, {remove_subscriptions, SubsList}, infinity). get_subscribers(User) -> gen_server:call(?SERVER, {get_subscribers, User}). %% init([]) -> ok = mnesia:start(), io:format("Waiting on mnesia tables..\n",[]), mnesia:wait_for_tables([subscription], 30000), Info = mnesia:table_info(subscription, all), io:format("OK. Subscription table info: \n~w\n\n",[Info]), {ok, #state{}}. handle_call({stop}, _From, State) -> {stop, stop, State}; handle_call({add_subscriptions, SubsList}, _From, State) -> % Transactionally is slower: % F = fun() -> % [ ok = mnesia:write(S) || S <- SubsList ] % end, % mnesia:transaction(F), [ mnesia:dirty_write(S) || S <- SubsList ], {reply, ok, State}; handle_call({remove_subscriptions, SubsList}, _From, State) -> F = fun() -> [ ok = mnesia:delete_object(S) || S <- SubsList ] end, mnesia:transaction(F), {reply, ok, State}; handle_call({get_subscribers, User}, From, State) -> F = fun() -> Subs = mnesia:dirty_match_object(#subscription{subscriber='_', subscribee=User}), Users = [Dude || #subscription{subscriber=Dude, subscribee=_} <- Subs], gen_server:reply(From, Users) end, spawn(F), {noreply, State}. handle_cast(_Msg, State) -> {noreply, State}. handle_info(_Msg, State) -> {noreply, State}. terminate(_Reason, _State) -> mnesia:stop(), ok. code_change(_OldVersion, State, _Extra) -> io:format("Reloading code for ?MODULE\n",[]), {ok, State}. %% first_run() -> mnesia:create_schema([node()]), ok = mnesia:start(), Ret = mnesia:create_table(subscription, [ {disc_copies, [node()]}, {attributes, record_info(fields, subscription)}, {index, [subscribee]}, %index subscribee too {type, bag} ]), Ret.

उल्लेखनीय:

• मैंने पूर्ण पथ का उपयोग करते हुए Mnesia के लिए आवश्यक qlc.hrl को शामिल किया। यह अच्छा नहीं है, लेकिन एक अलग तरीके से मैं सफल नहीं हुआ।

• get_subscribers एक अन्य प्रक्रिया को जन्म देता है और gen_server का उपयोग करके उसी प्रक्रिया के प्रति प्रतिक्रिया के निर्माण को दर्शाता है: उत्तर। इसका अर्थ है कि अगर हम अक्सर लुकअप कहते हैं तो gen_server लूप इस कॉल पर ब्लॉक नहीं होगा।

• आरआर ("सबमनगर.एलआर")। नीचे दिया गया उदाहरण आपको erl शेल में रिकॉर्ड परिभाषा का उपयोग करने की अनुमति देता है। Record.hrl फ़ाइल में अपनी परिभाषाएँ रखना और इसे अपने मॉड्यूल में शामिल करना सबसे अच्छी शैली है। मैंने शॉर्ट के लिए ऐसा किया।

अब जांच करें। first_run () Mnesia स्कीमा बनाता है, इसलिए इसे पहले कॉल करना महत्वपूर्ण है। मनेसिया के साथ एक और संभावित गड़बड़ यह है कि (डिफ़ॉल्ट रूप से) केवल नोड जो इसे बनाता है वह डेटाबेस तक पहुंच सकता है, इसलिए खोल को एक नाम दें।

 $ mkdir /var/mnesia $ erl -boot start_sasl -mnesia dir '"/var/mnesia_data"' -sname subsman (subsman@localhost)1> c(subsmanager). {ok,subsmanager} (subsman@localhost)2> subsmanager:first_run(). ... {atomic,ok} (subsman@localhost)3> subsmanager:start_link(). Waiting on mnesia tables.. OK. Subscription table info: [{access_mode,read_write},{active_replicas,[subsman@localhost]},{arity,3},{attributes,[subscriber,subscribee]},{checkpoints,[]},{commit_work,[{index,bag,[{3,{ram,57378}}]}]},{cookie,{{1224,800064,900003},subsman@localhost}},{cstruct,{cstruct,subscription,bag,[],[subsman@localhost],[],0,read_write,[3],[],false,subscription,[subscriber,subscribee],[],[],{{1224,863164,904753},subsman@localhost},{{2,0},[]}}},{disc_copies,[subsman@localhost]},{disc_only_copies,[]},{frag_properties,[]},{index,[3]},{load_by_force,false},{load_node,subsman@localhost},{load_order,0},{load_reason,{dumper,create_table}},{local_content,false},{master_nodes,[]},{memory,288},{ram_copies,[]},{record_name,subscription},{record_validation,{subscription,3,bag}},{type,bag},{size,0},{snmp,[]},{storage_type,disc_copies},{subscribers,[]},{user_properties,[]},{version,{{2,0},[]}},{where_to_commit,[{subsman@localhost,disc_copies}]},{where_to_read,subsman@localhost},{where_to_write,[subsman@localhost]},{wild_pattern,{subscription,'_','_'}},{{index,3},57378}] {ok,<0.105.0>} (subsman@localhost)4> rr("subsmanager.erl"). [state,subscription] (subsman@localhost)5> subsmanager:add_subscriptions([ #subscription{subscriber=alice, subscribee=rj} ]). ok (subsman@localhost)6> subsmanager:add_subscriptions([ #subscription{subscriber=bob, subscribee=rj} ]). ok (subsman@localhost)7> subsmanager:get_subscribers(rj). [bob,alice] (subsman@localhost)8> subsmanager:remove_subscriptions([ #subscription{subscriber=bob, subscribee=rj} ]). ok (subsman@localhost)8> subsmanager:get_subscribers(rj). [alice] (subsman@localhost)10> subsmanager:get_subscribers(charlie). []

हम पूर्णांक Ids का उपयोग उपयोगकर्ताओं के बीच अंतर करने के लिए करेंगे - लेकिन इस परीक्षण के लिए मैंने परमाणुओं (rj, alice, bob) का उपयोग किया और यह मान लिया कि alice और bob rj के मित्र हैं। यह बहुत अच्छा है कि मानेसिया (और ets / dets) परवाह नहीं करता है कि आपने किस प्रकार का उपयोग किया है - कोई भी इरलंग शब्द मान्य है। इसका मतलब है कि विभिन्न प्रकारों का समर्थन करने के लिए अद्यतन करना मुश्किल नहीं होगा।

राउटर बदलें

विशिष्ट उपयोगकर्ताओं को संदेश भेजने के बजाय, वह है: राउटर: भेजें (123, "हैलो उपयोगकर्ता 123"), हम संदेशों को "चिह्नित" करते हैं - संदेश उत्पन्न करने वाला व्यक्ति - और एक राउटर है जो प्रत्येक हस्ताक्षरित उपयोगकर्ता को संदेश भेजता है। दूसरे शब्दों में, एपीआई इस तरह से काम करेगा: राउटर: सेंड (123, "हेलो हर यूजर सब्स्क्राइब टू 123")।

 -module(router). -behaviour(gen_server). -export([start_link/0]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -export([send/2, login/2, logout/1]). -define(SERVER, global:whereis_name(?MODULE)). % will hold bidirectional mapping between id <–> pid -record(state, {pid2id, id2pid}). start_link() -> gen_server:start_link({global, ?MODULE}, ?MODULE, [], []). % sends Msg to anyone subscribed to Id send(Id, Msg) -> gen_server:call(?SERVER, {send, Id, Msg}). login(Id, Pid) when is_pid(Pid) -> gen_server:call(?SERVER, {login, Id, Pid}). logout(Pid) when is_pid(Pid) -> gen_server:call(?SERVER, {logout, Pid}). %% init([]) -> % set this so we can catch death of logged in pids: process_flag(trap_exit, true), % use ets for routing tables {ok, #state{ pid2id = ets:new(?MODULE, [bag]), id2pid = ets:new(?MODULE, [bag]) } }. handle_call({login, Id, Pid}, _From, State) when is_pid(Pid) -> ets:insert(State#state.pid2id, {Pid, Id}), ets:insert(State#state.id2pid, {Id, Pid}), link(Pid), % tell us if they exit, so we can log them out %io:format("~w logged in as ~w\n",[Pid, Id]), {reply, ok, State}; handle_call({logout, Pid}, _From, State) when is_pid(Pid) -> unlink(Pid), PidRows = ets:lookup(State#state.pid2id, Pid), case PidRows of [] -> ok; _ -> IdRows = [ {I,P} || {P,I} <- PidRows ], % invert tuples ets:delete(State#state.pid2id, Pid), % delete all pid->id entries [ ets:delete_object(State#state.id2pid, Obj) || Obj <- IdRows ] % and all id->pid end, %io:format("pid ~w logged out\n",[Pid]), {reply, ok, State}; handle_call({send, Id, Msg}, From, State) -> F = fun() -> % get users who are subscribed to Id: Users = subsmanager:get_subscribers(Id), io:format("Subscribers of ~w = ~w\n",[Id, Users]), % get pids of anyone logged in from Users list: Pids0 = lists:map( fun(U)-> [ P || { _I, P } <- ets:lookup(State#state.id2pid, U) ] end, [ Id | Users ] % we are always subscribed to ourselves ), Pids = lists:flatten(Pids0), io:format("Pids: ~w\n", [Pids]), % send Msg to them all M = {router_msg, Msg}, [ Pid ! M || Pid <- Pids ], % respond with how many users saw the message gen_server:reply(From, {ok, length(Pids)}) end, spawn(F), {noreply, State}. % handle death and cleanup of logged in processes handle_info(Info, State) -> case Info of {'EXIT', Pid, _Why} -> handle_call({logout, Pid}, blah, State); Wtf -> io:format("Caught unhandled message: ~w\n", [Wtf]) end, {noreply, State}. handle_cast(_Msg, State) -> {noreply, State}. terminate(_Reason, _State) -> ok. code_change(_OldVsn, State, _Extra) -> {ok, State}.

थोड़ा परीक्षण - मैंने आईडी के बजाय परमाणुओं का उपयोग किया।

 (subsman@localhost)1> c(subsmanager), c(router), rr("subsmanager.erl"). (subsman@localhost)2> subsmanager:start_link(). (subsman@localhost)3> router:start_link(). (subsman@localhost)4> Subs = [#subscription{subscriber=alice, subscribee=rj}, #subscription{subscriber=bob, subscribee=rj}]. [#subscription{subscriber = alice,subscribee = rj}, #subscription{subscriber = bob,subscribee = rj}] (subsman@localhost)5> subsmanager:add_subscriptions(Subs). ok (subsman@localhost)6> router:send(rj, "RJ did something"). Subscribers of rj = [bob,alice] Pids: [] {ok,0} (subsman@localhost)7> router:login(alice, self()). ok (subsman@localhost)8> router:send(rj, "RJ did something"). Subscribers of rj = [bob,alice] Pids: [<0.46.0>] {ok,1} (subsman@localhost)9> receive {router_msg, M} -> io:format("~s\n",[M]) end. RJ did something ok

यह देखा जा सकता है कि एलिस संदेश प्राप्त कर सकता है जब कोई व्यक्ति जिसे वह सदस्यता दे रहा है वह एक संदेश भेजता है, भले ही संदेश सीधे एलिस को नहीं भेजा गया हो। निष्कर्ष से पता चलता है कि राउटर ने संभावित लक्ष्यों की पहचान [alice, bob] के रूप में की, लेकिन एक व्यक्ति को, alice, क्योंकि bob को अधिकृत नहीं किया गया था, को संदेश भेज दिया।

एक साधारण डेटासेट बनाना

हम यादृच्छिक पर बहुत सारे संबंध उत्पन्न कर सकते हैं, लेकिन यह विशेष रूप से यथार्थवादी नहीं है। सामाजिक नेटवर्क में आमतौर पर कुछ सुपर लोकप्रिय उपयोगकर्ता होते हैं (कुछ ट्विटर उपयोगकर्ताओं के 100,000 से अधिक अनुयायी हैं), और बहुत से लोग केवल कुछ मुट्ठी भर दोस्त हैं।

डेटासेट उत्पन्न करने के लिए, मैंने उत्कृष्ट igraph लाइब्रेरी से पायथन मॉड्यूल का उपयोग किया:

 import igraph g = igraph.Graph.Barabasi(1000000, 15, directed=False) print "Edges: " + str(g.ecount()) + " Verticies: " + str(g.vcount()) g.write_edgelist("fakefriends.txt")

Mnesia में डेटा अपलोड करना

यह छोटा मॉड्यूल नकली गर्लफ्रेंड.टैक्स फाइल को पढ़ता है और सदस्यता की सूची बनाता है।

 -module(readfriends). -export([load/1]). -record(subscription, {subscriber, subscribee}). load(Filename) -> for_each_line_in_file(Filename, fun(Line, Acc) -> [As, Bs] = string:tokens(string:strip(Line, right, $\n), " "), {A, _} = string:to_integer(As), {B, _} = string:to_integer(Bs), [ #subscription{subscriber=A, subscribee=B} | Acc ] end, [read], []). % via: http://www.trapexit.org/Reading_Lines_from_a_File for_each_line_in_file(Name, Proc, Mode, Accum0) -> {ok, Device} = file:open(Name, Mode), for_each_line(Device, Proc, Accum0). for_each_line(Device, Proc, Accum) -> case io:get_line(Device, "") of eof -> file:close(Device), Accum; Line -> NewAccum = Proc(Line, Accum), for_each_line(Device, Proc, NewAccum) end.

अब सबसमैन शेल में आप एक टेक्स्ट फ़ाइल से पढ़ सकते हैं और सदस्यताएँ जोड़ सकते हैं:

 $ erl -name router@minifeeds4.gs2 +K true +A 128 -setcookie secretcookie -mnesia dump_log_write_threshold 50000 -mnesia dc_dump_limit 40 erl> c(readfriends), c(subsmanager). erl> subsmanager:first_run(). erl> subsmanager:start_link(). erl> subsmanager:add_subscriptions( readfriends:load("fakefriends.txt") ).

अतिरिक्त मापदंडों पर ध्यान दें - वे "** चेतावनी ** मानेसिया अतिभारित हैं" संदेशों से बचने में मदद करेंगे। मेन्सिया प्रलेखन में कई अन्य उल्लेखनीय सेटिंग्स शामिल हैं।

1000000

एक मेजबान से एक मिलियन टीसीपी कनेक्शन बनाना तुच्छ नहीं है। मुझे लगता है कि जो लोग नियमित रूप से बड़ी संख्या में यौगिकों के मॉडल के लिए समर्पित छोटे समूहों पर ऐसा करते हैं, वे शायद त्सुंग जैसे वास्तविक उपकरण का उपयोग करते हैं। यहां तक कि भाग 1 से विन्यास के साथ, हम अभी भी एक कठिन बंदरगाह सीमा में चलते हैं। Tcp कनेक्शन बनाते समय, क्लाइंट पोर्ट को / proc / sys / net / ipv4 / ip_local_port_range में रेंज से आवंटित किया जाता है। यदि आप इसे मैन्युअल रूप से सेट करते हैं या स्वचालित पोर्ट का उपयोग करते हैं, तो यह कोई बात नहीं है। भाग 1 में, हमने "1024 65535" पर सीमा तय की, अर्थात हमारे पास 65535-1024 = 64511 अनपेक्षित उपलब्ध पोर्ट हैं। उनमें से कुछ का उपयोग अन्य प्रक्रियाओं द्वारा किया जाएगा, लेकिन हम कभी भी 64511 से अधिक ग्राहकों को पास नहीं करेंगे, क्योंकि हम बंदरगाहों से बाहर भाग जाएंगे।

स्थानीय पोर्ट रेंज आईपी से जुड़ा है, इसलिए यदि हम विभिन्न स्थानीय आईपी पते से आउटगोइंग कनेक्शन बनाते हैं, तो हम 64511 से अधिक आउटगोइंग कनेक्शन खोल पाएंगे।

तो आइए प्रत्येक से 62,000 कनेक्शन बनाने के लिए 17 नए आईपी पते बनाएं - यह हमें कुल 1,054,000 कनेक्शन देगा:

 $ for i in `seq 1 17`; do echo sudo ifconfig eth0:$i 10.0.0.$i up ; done

यदि आप ifconfig को अभी चेक करते हैं, तो आपको अपना वर्चुअल इंटरफेस देखना चाहिए: eth0: 1, eth0: 2 ... eth0: 17, प्रत्येक एक अलग IP पते के साथ।

स्थानीय आईपी का चयन करने के लिए अब जो कुछ बचता है वह भाग 1 से बाढ़ को बदलना है। दुर्भाग्य से, erlang http क्लाइंट आपको स्रोत IP निर्धारित करने की अनुमति नहीं देता है।

इस बिंदु पर, मैं एक और संभावना पर विचार कर रहा था: 17 आईपी जोड़े का उपयोग करने के लिए - एक सर्वर पर और एक क्लाइंट पर - प्रत्येक जोड़ी अपने स्वयं के पृथक / 30 सबनेट में। मुझे लगता है कि अगर मैंने क्लाइंट को किसी भी आईपी सर्वर से कनेक्ट करने के लिए मजबूर किया, तो यह स्थानीय पते को जोड़ी के दूसरे होने के लिए मजबूर करेगा, क्योंकि स्थानीय आईपी में से केवल एक ही इस सबनेट पर आईपी सर्वर तक पहुंचने में सक्षम होगा। सिद्धांत रूप में, इसका मतलब होगा कि ग्राहक मशीन पर एक स्थानीय स्रोत आईपी आवश्यक नहीं होगा (हालांकि सर्वर के आईपी पते की सीमा को परिभाषित करना होगा)। मुझे नहीं पता कि यह काम करेगा या नहीं - इस समय इसकी संभावना थी। अंत में, मैंने फैसला किया कि यह बहुत विकृत होगा।

gen_tcp आपको स्रोत पता निर्दिष्ट करने की अनुमति देता है, इसलिए मैंने एक कच्चे ग्राहक का उपयोग करके समाप्त किया:

 -module(floodtest2). -compile(export_all). -define(SERVERADDR, "10.1.2.3"). % where mochiweb is running -define(SERVERPORT, 8000). % Generate the config in bash like so (chose some available address space): % EACH=62000; for i in `seq 1 17`; do echo "{{10,0,0,$i}, $((($i-1)*$EACH+1)), $(($i*$EACH))}, "; done run(Interval) -> Config = [ {{10,0,0,1}, 1, 62000}, {{10,0,0,2}, 62001, 124000}, {{10,0,0,3}, 124001, 186000}, {{10,0,0,4}, 186001, 248000}, {{10,0,0,5}, 248001, 310000}, {{10,0,0,6}, 310001, 372000}, {{10,0,0,7}, 372001, 434000}, {{10,0,0,8}, 434001, 496000}, {{10,0,0,9}, 496001, 558000}, {{10,0,0,10}, 558001, 620000}, {{10,0,0,11}, 620001, 682000}, {{10,0,0,12}, 682001, 744000}, {{10,0,0,13}, 744001, 806000}, {{10,0,0,14}, 806001, 868000}, {{10,0,0,15}, 868001, 930000}, {{10,0,0,16}, 930001, 992000}, {{10,0,0,17}, 992001, 1054000}], start(Config, Interval). start(Config, Interval) -> Monitor = monitor(), AdjustedInterval = Interval / length(Config), [ spawn(fun start/5, [Lower, Upper, Ip, AdjustedInterval, Monitor]) || {Ip, Lower, Upper} <- Config ], ok. start(LowerID, UpperID, _, _, _) when LowerID == UpperID -> done; start(LowerID, UpperID, LocalIP, Interval, Monitor) -> spawn(fun connect/5, [?SERVERADDR, ?SERVERPORT, LocalIP, "/test/"++LowerID, Monitor]), receive after Interval -> start(LowerID + 1, UpperID, LocalIP, Interval, Monitor) end. connect(ServerAddr, ServerPort, ClientIP, Path, Monitor) -> Opts = [binary, {packet, 0}, {ip, ClientIP}, {reuseaddr, true}, {active, false}], {ok, Sock} = gen_tcp:connect(ServerAddr, ServerPort, Opts), Monitor ! open, ReqL = io_lib:format("GET ~s\r\nHost: ~s\r\n\r\n", [Path, ServerAddr]), Req = list_to_binary(ReqL), ok = gen_tcp:send(Sock, [Req]), do_recv(Sock, Monitor), (catch gen_tcp:close(Sock)), ok. do_recv(Sock, Monitor)-> case gen_tcp:recv(Sock, 0) of {ok, B} -> Monitor ! {bytes, size(B)}, io:format("Recvd ~s\n", [ binary_to_list(B)]), io:format("Recvd ~w bytes\n", [size(B)]), do_recv(Sock, Monitor); {error, closed} -> Monitor ! closed, closed; Other -> Monitor ! closed, io:format("Other:~w\n",[Other]) end. % Monitor process receives stats and reports how much data we received etc: monitor() -> Pid = spawn(?MODULE, monitor0, [{0,0,0,0}]), timer:send_interval(10000, Pid, report), Pid. monitor0({Open, Closed, Chunks, Bytes}=S) -> receive report -> io:format("{Open, Closed, Chunks, Bytes} = ~w\n",[S]); open -> monitor0({Open + 1, Closed, Chunks, Bytes}); closed -> monitor0({Open, Closed + 1, Chunks, Bytes}); chunk -> monitor0({Open, Closed, Chunks + 1, Bytes}); {bytes, B} -> monitor0({Open, Closed, Chunks, Bytes + B}) end.

सबसे पहले, मैं भाग 1 से मोचीवेब एप्लिकेशन से जुड़ा - यह सिर्फ प्रत्येक क्लाइंट को प्रत्येक 10 सेकंड में एक संदेश भेजता है।

 erl> c(floodtest2), floodtest2:run(20).

यह जल्दी से मेरी सारी मेमोरी खा गया

यह पता चला है कि gen_tcp के साथ बड़ी संख्या में कनेक्शन खोलने से बहुत सारी मेमोरी खत्म हो जाती है। मुझे लगता है कि यह काम करने के लिए ~ 36 जीबी ले जाएगा। मुझे अपने erlang http क्लाइंट को ऑप्टिमाइज़ करने की कोशिश में कोई दिलचस्पी नहीं थी, और 32GB से अधिक मेमोरी वाली एकमात्र मशीन जो मुझे मिल सकती थी, वह हमारे डेटाबेस में से एक थी, और मुझे Last.fm को बंद करने का एक अच्छा बहाना नहीं मिला, जबकि मैं आसपास गड़बड़ कर रहा था :)

इस बिंदु पर, मैंने कोशिश की और परीक्षण किए गए परिवाद को याद करने का फैसला किया, जिसमें एक HTTP एपीआई है। नए संस्करणों में http एपीआई में evhttp_connection_set_local_address फ़ंक्शन भी है।

यहाँ सी http क्लाइंट का उपयोग कर रहा है libevent:

 #include <sys/types.h> #include <sys/time.h> #include <sys/queue.h> #include <stdlib.h> #include <err.h> #include <event.h> #include <evhttp.h> #include <unistd.h> #include <stdio.h> #include <sys/socket.h> #include <netinet/in.h> #include <time.h> #include <pthread.h> #define BUFSIZE 4096 #define NUMCONNS 62000 #define SERVERADDR "10.103.1.43" #define SERVERPORT 8000 #define SLEEP_MS 10 char buf[BUFSIZE]; int bytes_recvd = 0; int chunks_recvd = 0; int closed = 0; int connected = 0; // called per chunk received void chunkcb(struct evhttp_request * req, void * arg) { int s = evbuffer_remove( req->input_buffer, &buf, BUFSIZE ); //printf("Read %d bytes: %s\n", s, &buf); bytes_recvd += s; chunks_recvd++; if(connected >= NUMCONNS && chunks_recvd%10000==0) printf(">Chunks: %d\tBytes: %d\tClosed: %d\n", chunks_recvd, bytes_recvd, closed); } // gets called when request completes void reqcb(struct evhttp_request * req, void * arg) { closed++; } int main(int argc, char **argv) { event_init(); struct evhttp *evhttp_connection; struct evhttp_request *evhttp_request; char addr[16]; char path[32]; // eg: "/test/123" int i,octet; for(octet=1; octet<=17; octet++){ sprintf(&addr, "10.224.0.%d", octet); for(i=1;i<=NUMCONNS;i++) { evhttp_connection = evhttp_connection_new(SERVERADDR, SERVERPORT); evhttp_connection_set_local_address(evhttp_connection, &addr); evhttp_set_timeout(evhttp_connection, 864000); // 10 day timeout evhttp_request = evhttp_request_new(reqcb, NULL); evhttp_request->chunk_cb = chunkcb; sprintf(&path, "/test/%d", ++connected); if(i%100==0) printf("Req: %s\t->\t%s\n", addr, &path); evhttp_make_request( evhttp_connection, evhttp_request, EVHTTP_REQ_GET, path ); evhttp_connection_set_timeout(evhttp_request->evcon, 864000); event_loop( EVLOOP_NONBLOCK ); if( connected % 200 == 0 ) printf("\nChunks: %d\tBytes: %d\tClosed: %d\n", chunks_recvd, bytes_recvd, closed); usleep(SLEEP_MS*1000); } } event_dispatch(); return 0; }

अधिकांश मापदंडों को #define के रूप में कोडित किया जाता है, इसलिए आप इसे संपादित कर सकते हैं और पुन: प्रस्तुत कर सकते हैं:

 $ gcc -o httpclient httpclient.c -levent $ ./httpclient

यह अभी भी 64,500 से अधिक बंदरगाहों को खोलने में असमर्थ है।

64,500 से अधिक कनेक्शन खोलने के लिए, आपको स्थानीय पता और स्थानीय पोर्ट को स्वयं निर्धारित करना होगा, और उनके अनुसार प्रबंधन करना होगा।

दुर्भाग्य से, libevent HTTP API में एक स्थानीय पोर्ट निर्दिष्ट करने का विकल्प नहीं है। मैं इस तरह एक समारोह जोड़ने के लिए libevent तय:

 void evhttp_connection_set_local_port(struct evhttp_connection *evcon, u_short port);.

यह एक आश्चर्यजनक सुखद अनुभव था: कामचलाऊ अच्छी तरह से लिखा हुआ लगता है, और प्रलेखन बहुत अच्छा है।

इस संशोधित परिवाद के साथ, मैं उपरोक्त कोड में निम्नलिखित जोड़ने में सक्षम था:

 evhttp_connection_set_local_port(evhttp_connection, 1024+i);

अब विभिन्न पतों से कई कनेक्शन स्थानीय पते के लिए परिभाषित एक ही स्थानीय पोर्ट नंबर का उपयोग करने में सक्षम थे। मैंने क्लाइंट को पुन: स्थापित किया, और यह सुनिश्चित करने के लिए थोड़ी देर के लिए काम करने दिया कि यह बाधा को पार करेगा।

Netstat इसकी पुष्टि करता है:

 # netstat -n | awk '/^tcp/ {t[$NF]++}END{for(state in t){print state, t[state]}}' TIME_WAIT 8 ESTABLISHED 118222

इससे पता चलता है कि विभिन्न राज्यों में कितने बंदरगाह खुले हैं। हम आखिरकार 2 ^ 16 से अधिक कनेक्शन खोलने में कामयाब रहे।

अब हमारे पास एक उपकरण है जो एक एकल कंप्यूटर से एक लाख http कनेक्शन खोलने में सक्षम है। ऐसा लगता है कि यह प्रत्येक कनेक्शन के लिए लगभग 2 KB का उपयोग करता है, साथ ही कर्नेल क्या रखता है। यह हमारे मोचीवेब सर्वर का परीक्षण करने का समय है।

C1024K

इस परीक्षण के लिए, मैंने 4 विभिन्न सर्वरों का उपयोग किया। इस परीक्षण और पिछले वाले के बीच मुख्य अंतर सी में लिखा गया एक संशोधित ग्राहक है।

सर्वर 1 - क्वाड-कोर 2GHz CPU, 16GB RAM

• उपप्रबंधक लॉन्च करें

• डेटा लोड हो रहा है

• राउटर लॉन्च

सर्वर 2 - दोहरी क्वाड-कोर 2.8GHz सीपीयू, 32 जीबी रैम

• लॉन्च Mochiweb आवेदन

सर्वर 3 - क्वाड-कोर 2GHz CPU, 16GB RAM

• 17 आभासी आईपी पतों का निर्माण

• परिवाद स्थापित करें

• क्लाइंट स्टार्टअप: ./httpclient (प्रति सेकंड 100 कनेक्शन)

सर्वर 4 - डुअल-कोर 2GHz, 2GB रैम

• संदेश का एक गुच्छा भेजने के लिए msggen चल रहा है

कनेक्शन खोलने के दौरान और कुछ समय के लिए मेमोरी का उपयोग:

HttpClient में कनेक्शन के बीच अंतर्निहित 10ms देरी है, इसलिए एक लाख कनेक्शन खोलने में लगभग 3 घंटे लगते हैं। इसमें लगभग 25GB मेमोरी लगी थी। यहाँ गंगालिया की नज़र से मेरा सर्वर कैसा दिखता है:

आप देख सकते हैं कि इसमें लगभग 38GB लगते हैं, और फिर स्वैपिंग शुरू होती है। मुझे संदेह है कि मुख्य रूप से मुख्य खपत में अंतर है।

संदेशों का औसत 1000 प्रक्रियाओं का उपयोग करके उत्पन्न किया गया था, प्रति 60ms प्रति प्रक्रिया के संदेशों के बीच औसत समय, प्रति सेकंड लगभग 16666 संदेश देता है:

 erl> [ spawn( fun()->msggen:start(1000000, 10+random:uniform(100), 1000000) end) || I <- lists:seq(1,1000) ].

गंगालिया में सर्वर 4:

लगभग 10 एमबी प्रति सेकंड - 16.666 संदेश।

जब मैंने संदेश भेजने शुरू किए, तो पहले सर्वर पर लोड कम रहा। दूसरे सर्वर पर सीपीयू की खपत बढ़ी:

स्वाभाविक रूप से, क्योंकि संदेशों को संसाधित करने के लिए हाइबरनेट () से बाहर निकलने की प्रक्रिया, मेमोरी का उपयोग थोड़ा बढ़ जाता है। किसी भी संदेश के बिना सभी खुले कनेक्शन होने से मेमोरी उपयोग में इष्टतम है। आश्चर्य नहीं कि किसी भी क्रिया के लिए अधिक स्मृति की आवश्यकता होती है।

तो स्मृति कहाँ बहती है? मोचीव्ब को 1,000,000 सक्रिय कनेक्शनों को खुला रखने के लिए 40 जीबी रैम की आवश्यकता होती है। लोड के तहत, 30 जीबी तक मेमोरी का उपयोग मोचीवेब और शेष 10 जीबी कोर द्वारा किया जाएगा। दूसरे शब्दों में, प्रति कनेक्शन लगभग 40Kb की जरूरत है।

बहुत सारे कनेक्शन के साथ विभिन्न परीक्षणों में, मैंने sysctl.conf में कुछ अतिरिक्त बदलाव किए। मैं परीक्षण और त्रुटि के साथ आया था, और मुझे नहीं पता कि मूल्यों को क्या बदलना है। मेरी नीति यह थी कि मैं /var/log/kern.log की जांच करने के लिए एक त्रुटि की प्रतीक्षा करूं और देखूं कि रहस्यमय त्रुटि मुझे क्या बताती है। यहाँ सेटिंग्स हैं:

 net.core.rmem_max = 33554432 net.core.wmem_max = 33554432 net.ipv4.tcp_rmem = 4096 16384 33554432 net.ipv4.tcp_wmem = 4096 16384 33554432 net.ipv4.tcp_mem = 786432 1048576 26777216 net.ipv4.tcp_max_tw_buckets = 360000 net.core.netdev_max_backlog = 2500 vm.min_free_kbytes = 65536 vm.swappiness = 0 net.ipv4.ip_local_port_range = 1024 65535

मैं अधिक सूचित स्पष्टीकरण बनाने के लिए लिनक्स टीसीपी सेटिंग्स के बारे में अधिक जानना चाहूंगा। लगभग निश्चित रूप से, ये सेटिंग्स इष्टतम नहीं हैं, लेकिन कम से कम यह 1,000,000 कनेक्शन तक पहुंचने के लिए पर्याप्त था।

लिबवेंट पर एरलंग नोड

Libevent के लिए HTTP API के साथ छेड़छाड़ करने के बाद, उपरोक्त परीक्षण को libevent HTTPd के साथ N में लिखा जाना पूरी तरह से उचित लगता है।

मैं erlang पर अधिक से अधिक कोड छोड़ना चाहूंगा, तो आइए C - C में एक न्यूनतम काम करते हैं - केवल HTTP कनेक्शन को संभालते हुए।

लिबवेंट में एक एसिंक्रोनस HTTP एपीआई है, जो HTTP सर्वर के कार्यान्वयन को तुच्छ बनाता है। मैंने एरलंग के लिए सी इंटरफेस की कोशिश करने के लिए एक कारण भी देखा। यह एक HTTP सर्वर है जो कि libevent का उपयोग करता है, जो पूरे Id (जैसे हमारे mochiweb एप्लिकेशन) का उपयोग करके उपयोगकर्ताओं की पहचान करता है, और एक Erlang C-नोड के रूप में भी कार्य करता है।

यह एर्लांग-नामित होस्ट से जुड़ता है, {123, << "हैलो उपयोगकर्ता 123" >>} जैसे संदेशों को सुनता है, अगर यह जुड़ा हुआ है, तो उपयोगकर्ता 123 को "हैलो उपयोगकर्ता 123" भेजता है। असंबद्ध उपयोगकर्ताओं के लिए संदेश पहले की तरह ही छोड़ दिए जाते हैं।

 #include <sys/types.h> #include <sys/time.h> #include <sys/queue.h> #include <stdlib.h> #include <err.h> #include <event.h> #include <evhttp.h> #include <stdio.h> #include <sys/socket.h> #include <netinet/in.h> #include "erl_interface.h" #include "ei.h" #include <pthread.h> #define BUFSIZE 1024 #define MAXUSERS (17*65536) // C1024K // List of current http requests by uid: struct evhttp_request * clients[MAXUSERS+1]; // Memory to store uids passed to the cleanup callback: int slots[MAXUSERS+1]; // called when user disconnects void cleanup(struct evhttp_connection *evcon, void *arg) { int *uidp = (int *) arg; fprintf(stderr, "disconnected uid %d\n", *uidp); clients[*uidp] = NULL; } // handles http connections, sets them up for chunked transfer, // extracts the user id and registers in the global connection table, // also sends a welcome chunk. void request_handler(struct evhttp_request *req, void *arg) { struct evbuffer *buf; buf = evbuffer_new(); if (buf == NULL){ err(1, "failed to create response buffer"); } evhttp_add_header(req->output_headers, "Content-Type", "text/html; charset=utf-8"); int uid = -1; if(strncmp(evhttp_request_uri(req), "/test/", 6) == 0){ uid = atoi( 6+evhttp_request_uri(req) ); } if(uid <= 0){ evbuffer_add_printf(buf, "User id not found, try /test/123 instead"); evhttp_send_reply(req, HTTP_NOTFOUND, "Not Found", buf); evbuffer_free(buf); return; } if(uid > MAXUSERS){ evbuffer_add_printf(buf, "Max uid allowed is %d", MAXUSERS); evhttp_send_reply(req, HTTP_SERVUNAVAIL, "We ran out of numbers", buf); evbuffer_free(buf); return; } evhttp_send_reply_start(req, HTTP_OK, "OK"); // Send welcome chunk: evbuffer_add_printf(buf, "Welcome, Url: '%s' Id: %d\n", evhttp_request_uri(req), uid); evhttp_send_reply_chunk(req, buf); evbuffer_free(buf); // put reference into global uid->connection table: clients[uid] = req; // set close callback evhttp_connection_set_closecb( req->evcon, cleanup, &slots[uid] ); } // runs in a thread – the erlang c-node stuff // expects msgs like {uid, msg} and sends aa 'msg' chunk to uid if connected void cnode_run() { int fd; /* fd to Erlang node */ int got; /* Result of receive */ unsigned char buf[BUFSIZE]; /* Buffer for incoming message */ ErlMessage emsg; /* Incoming message */ ETERM *uid, *msg; erl_init(NULL, 0); if (erl_connect_init(1, "secretcookie", 0) == -1) erl_err_quit("erl_connect_init"); if ((fd = erl_connect("httpdmaster@localhost")) < 0) erl_err_quit("erl_connect"); fprintf(stderr, "Connected to httpdmaster@localhost\n\r"); struct evbuffer *evbuf; while (1) { got = erl_receive_msg(fd, buf, BUFSIZE, &emsg); if (got == ERL_TICK) { continue; } else if (got == ERL_ERROR) { fprintf(stderr, "ERL_ERROR from erl_receive_msg.\n"); break; } else { if (emsg.type == ERL_REG_SEND) { // get uid and body data from eg: {123, <<"Hello">>} uid = erl_element(1, emsg.msg); msg = erl_element(2, emsg.msg); int userid = ERL_INT_VALUE(uid); char *body = (char *) ERL_BIN_PTR(msg); int body_len = ERL_BIN_SIZE(msg); // Is this userid connected? if(clients[userid]){ fprintf(stderr, "Sending %d bytes to uid %d\n", body_len, userid); evbuf = evbuffer_new(); evbuffer_add(evbuf, (const void*)body, (size_t) body_len); evhttp_send_reply_chunk(clients[userid], evbuf); evbuffer_free(evbuf); }else{ fprintf(stderr, "Discarding %d bytes to uid %d – user not connected\n", body_len, userid); // noop } erl_free_term(emsg.msg); erl_free_term(uid); erl_free_term(msg); } } } // if we got here, erlang connection died. // this thread is supposed to run forever // TODO – gracefully handle failure / reconnect / etc pthread_exit(0); } int main(int argc, char **argv) { // Launch the thread that runs the cnode: pthread_attr_t tattr; pthread_t helper; int status; pthread_create(&helper, NULL, cnode_run, NULL); int i; for(i=0;i<=MAXUSERS;i++) slots[i]=i; // Launch libevent httpd: struct evhttp *httpd; event_init(); httpd = evhttp_start("0.0.0.0", 8000); evhttp_set_gencb(httpd, request_handler, NULL); event_dispatch(); // Not reached, event_dispatch() shouldn't return evhttp_free(httpd); return 0; }

उपयोगकर्ताओं की अधिकतम संख्या #define का उपयोग करके सेट की गई है, पोर्ट 8000 सुनी जाती है और उपयोगकर्ताओं से अपेक्षा की जाती है। कुकी नोड का Erlang नाम हार्ड-कोडेड भी है।

वह नोड प्रारंभ करें जिससे सर्वर कनेक्ट होगा:

 $ erl -setcookie secretcookie -sname httpdmaster@localhost

C नोड चलाएँ:

 $ gcc -o httpdcnode httpdcnode.c -lerl_interface -lei -levent $ ./httpdcnode

सत्यापित करें कि नोड दिखाई दे रहा है:

 erl> nodes(hidden). [c1@localhost]

ओपन लोकलहोस्ट : 8000 / टेस्ट / 123 अपने ब्राउज़र में । आपको एक स्वागत संदेश देखना चाहिए।

अगला, C नोड को एक संदेश भेजें:

 erl> {any, c1@localhost} ! {123, <<"Hello Libevent World">>}.

कृपया ध्यान दें कि हम Pid का उपयोग नहीं करते हैं - हम {procname, node} का उपयोग करते हैं। हम "किसी भी" का उपयोग एक प्रक्रिया के नाम के रूप में करते हैं जिसे सी-नोड द्वारा अनदेखा किया जाता है।

अब आप Erlang के माध्यम से संदेश देने में सक्षम हैं, लेकिन सभी कनेक्शनों को libevent C प्रोग्राम द्वारा नियंत्रित किया जाता है, जो Erlang नोड के रूप में कार्य करता है।

डिबग आउटपुट को हटाने के बाद, मैंने 1,000,000 क्लाइंट को उसी क्लाइंट का उपयोग करके httpdcnode सर्वर से जोड़ा है जो ऊपर के समान क्लाइंट का उपयोग कर रहा है। मशीन ने कुल 10 जीबी मेमोरी का इस्तेमाल किया।

सर्वर मेमोरी 2GB के आसपास स्थिर थी:

प्रति कनेक्शन 2KB के बारे में।

मेमोरी स्थिति:

 Mem: 32968672k total, 9636488k used, 23332184k free, 180k buffers

तो कर्नेल / टीसीपी स्टैक एक अतिरिक्त 8Kb प्रति कनेक्शन लेता है, जो उच्च लगता है, लेकिन मेरे पास तुलना करने के लिए कुछ भी नहीं है।

इस libevent-cnode सर्वर को स्वयं कुछ अतिरिक्त काम करने की आवश्यकता होती है। यह एक उपयोगकर्ता से कई कनेक्शन को संभालने के लिए बुद्धिमान नहीं है, और अभी भी "दौड़ की स्थिति" हैं यदि आप केवल संदेश भेजे जाने पर डिस्कनेक्ट करते हैं।

इसके बावजूद, हमने सभी दिलचस्प चीजों के लिए एर्लैंग का उपयोग किया, और निम्न स्तर के संचालन के लिए C + libevent का उपयोग किया गया। एक सी प्रोग्राम को ड्राइवर या सी नोड के रूप में चलाया जा सकता है, और एरलैंग इंटरफेस आपको एक सभ्य एपीआई दे सकता है। मुझे अभी भी इसके साथ प्रयोग करने की इच्छा है।

अंतिम निष्कर्ष

मेरे पास यह निर्धारित करने के लिए पर्याप्त डेटा है कि यदि हम एक बड़े लास्ट.फैम स्केल सिस्टम को तैनात करते हैं तो कौन से उपकरणों की आवश्यकता होगी। सबसे खराब स्थिति में भी, प्रति कनेक्शन 40Kb एक अच्छा परिणाम है - स्मृति इस समय बहुत सस्ती है, और लाखों उपयोगकर्ताओं का समर्थन करने के लिए 40GB एक समस्या नहीं है। 10GB भी बेहतर है।

धूमकेतु - 1,000,000 उपयोगकर्ताओं के भार के साथ मोचीवेब के लिए आवेदन। भाग 3/3