ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-MP/MP/Global.pm
(Generate patch)

Comparing AnyEvent-MP/MP/Global.pm (file contents):
Revision 1.50 by root, Sat Mar 3 13:07:19 2012 UTC vs.
Revision 1.51 by root, Sat Mar 3 19:43:41 2012 UTC

… …
61 61
62our $SRCNODE; 62our $SRCNODE;
63our %NODE_REQ; 63our %NODE_REQ;
64 64
65# only in global code 65# only in global code
66our %GLOBAL_MON; # monitors {family}{"" or key} 66our %GLOBAL_MON; # monitors {family}
67 67
68sub other_globals() { 68sub other_globals() {
69 grep $_ ne $NODE && node_is_up $_, keys %{ $GLOBAL_DB{"'g"} } 69 grep $_ ne $NODE && node_is_up $_, keys %{ $GLOBAL_DB{"'g"} }
70} 70}
71 71
… …
73sub g_broadcast { 73sub g_broadcast {
74 snd $_, @_ 74 snd $_, @_
75 for other_globals; 75 for other_globals;
76} 76}
77 77
78sub g_mon_check {
79 warn "g_mon_check<@_>\n";#d#
80
81 my %node = (
82 %{ $GLOBAL_MON{$_[1]}{$_[2]} },
83 %{ $GLOBAL_MON{$_[1]}{"" } },
84 %{ $GLOBAL_MON{"" }{"" } },
85 );
86
87 snd $_ => g_chg2 => $_[1], $_[2], @_ > 2 ? $_[3] : ()
88 for keys %node;
89}
90
91# add/replace a key in the database 78# add/replace a key in the database
92sub g_add($$$$) { 79sub g_add($$$$) {
93 $LOCAL_DBS{$_[0]}{$_[1]}{$_[2]} = 80 $LOCAL_DBS{$_[0]}{$_[1]}{$_[2]} =
94 $GLOBAL_DB {$_[1]}{$_[2]} = $_[3]; 81 $GLOBAL_DB {$_[1]}{$_[2]} = $_[3];
95 82
96 g_broadcast g_add => $_[1] => $_[2] => $_[3] 83 g_broadcast g_add => $_[1] => $_[2] => $_[3]
97 if exists $GLOBAL_SLAVE{$_[0]}; 84 if exists $GLOBAL_SLAVE{$_[0]};
98 85
99 warn "g_add<@_>\n";#d# 86 # tell subscribers we have added or replaced a key
100 &g_mon_check; 87 snd $_ => g_chg2 => $_[1], $_[2], $_[3]
88 for keys %{ $GLOBAL_MON{$_[1]} };
101} 89}
102 90
103# delete a key from the database 91# delete a key from the database
104sub g_del($$$) { 92sub g_del($$$) {
105 delete $LOCAL_DBS{$_[0]}{$_[1]}{$_[2]}; 93 delete $LOCAL_DBS{$_[0]}{$_[1]}{$_[2]};
94 delete $LOCAL_DBS{$_[0]}{$_[1]} unless %{ $LOCAL_DBS{$_[0]}{$_[1]} };
106 95
107 g_broadcast g_del => $_[1] => $_[2] 96 g_broadcast g_del => $_[1] => $_[2]
108 if exists $GLOBAL_SLAVE{$_[0]}; 97 if exists $GLOBAL_SLAVE{$_[0]};
109 98
110 delete $GLOBAL_DB{$_[1]}{$_[2]}; 99 delete $GLOBAL_DB{$_[1]}{$_[2]};
100 delete $GLOBAL_DB{$_[1]} unless %{ $GLOBAL_DB{$_[1]} }; # could be moved below for
111 101
112 # check if other node maybe still has the key, then we don't delete, but add 102 # check if other node maybe still has the key, then we don't delete, but add
113 for (values %LOCAL_DBS) { 103 for (values %LOCAL_DBS) {
114 if (exists $_->{$_[1]}{$_[2]}) { 104 if (exists $_->{$_[1]}{$_[2]}) {
115 $GLOBAL_DB{$_[1]}{$_[2]} = $_->{$_[1]}{$_[2]}; 105 my $val = $GLOBAL_DB{$_[1]}{$_[2]} = $_->{$_[1]}{$_[2]};
106
107 # tell subscribers we have added or replaced a key
108 snd $_ => g_chg2 => $_[1], $_[2], $val
109 for keys %{ $GLOBAL_MON{$_[1]} };
110
116 last; 111 last;
117 } 112 }
118 } 113 }
119 114
120 warn "g_del<@_>\n";#d# 115 # tell subscribers key is actually gone
121 &g_mon_check; 116 snd $_ => g_chg2 => $_[1], $_[2]
117 for keys %{ $GLOBAL_MON{$_[1]} };
118}
119
120# set the whole (node-local) database - previous value must be empty
121sub g_set($$) {
122 my ($node, $db) = @_;
123
124 while (my ($f, $k) = each %$db) {
125 g_add $node, $f => $_ => delete $k->{$_}
126 for keys %$k;
127 }
122} 128}
123 129
124# delete all keys from a database 130# delete all keys from a database
125sub g_clr($) { 131sub g_clr($) {
126 my ($node) = @_; 132 my ($node) = @_;
… …
132 } 138 }
133 139
134 delete $LOCAL_DBS{$node}; 140 delete $LOCAL_DBS{$node};
135} 141}
136 142
137# set the whole (node-local) database - previous value must be empty
138sub g_set($$) {
139 my ($node, $db) = @_;
140
141 while (my ($f, $k) = each %$db) {
142 g_add $node, $f => $_ => delete $k->{$_}
143 for keys %$k;
144 }
145}
146
147# gather node databases from slaves 143# gather node databases from slaves
148 144
149# other node wants to make us the master 145# other node wants to make us the master
150$NODE_REQ{g_slave} = sub { 146$NODE_REQ{g_slave} = sub {
151 my ($db) = @_; 147 my ($db) = @_;
… …
182 178
183 if (my $mon = delete $GLOBAL_SLAVE{$node}) { 179 if (my $mon = delete $GLOBAL_SLAVE{$node}) {
184 while (my ($f, $fv) = each %$mon) { 180 while (my ($f, $fv) = each %$mon) {
185 delete $GLOBAL_MON{$f}{$_} 181 delete $GLOBAL_MON{$f}{$_}
186 for keys %$fv; 182 for keys %$fv;
183
184 delete $GLOBAL_MON{$f}
185 unless %{ $GLOBAL_MON{$f} };
187 } 186 }
188 } 187 }
189} 188}
190 189
191# g_mon0 family key - stop monitoring 190# g_mon0 family - stop monitoring
192$NODE_REQ{g_mon0} = sub { 191$NODE_REQ{g_mon0} = sub {
193 delete $GLOBAL_MON{$_[0]}{$_[1]}{$SRCNODE->{id}}; 192 delete $GLOBAL_MON{$_[0]}{$SRCNODE->{id}};
193 delete $GLOBAL_MON{$_[0]} unless %{ $GLOBAL_MON{$_[0]} };
194
194 delete $GLOBAL_SLAVE{$SRCNODE->{id}}{$_[0]}{$_[1]}; 195 delete $GLOBAL_SLAVE{$SRCNODE->{id}}{$_[0]};
195}; 196};
196 197
197# g_mon1 family key - start monitoring 198# g_mon1 family key - start monitoring
198$NODE_REQ{g_mon1} = sub { 199$NODE_REQ{g_mon1} = sub {
199 undef $GLOBAL_SLAVE{$SRCNODE->{id}}{$_[0]}{$_[1]}; 200 undef $GLOBAL_SLAVE{$SRCNODE->{id}}{$_[0]};
200 undef $GLOBAL_MON{$_[0]}{$_[1]}{$SRCNODE->{id}}; 201 undef $GLOBAL_MON{$_[0]}{$SRCNODE->{id}};
201 202
202 snd $SRCNODE->{id}, g_chg1 => $_[0], $_[1], 203 snd $SRCNODE->{id}, g_chg1 => $_[0], $GLOBAL_DB{$_[0]};
203 defined $_[1] ? $GLOBAL_DB{$_[0]}{$_[1]} : $GLOBAL_DB{$_[0]};
204}; 204};
205 205
206############################################################################# 206#############################################################################
207# switch to global mode 207# switch to global mode
208 208

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines