File Coverage

blib/lib/MogileFS/Worker/JobMaster.pm
Criterion Covered Total %
statement 30 165 18.1
branch 0 52 0.0
condition 0 21 0.0
subroutine 10 23 43.4
pod 0 5 0.0
total 40 266 15.0


line stmt bran cond sub pod time code
1             package MogileFS::Worker::JobMaster;
2             # manages/monitors the internal queues for various workers.
3             # decided to have one of these per tracker instead of have workers
4             # all elect one per job type... should be able to reuse more code, and avoid
5             # relying on too many database locks.
6              
7 21     21   132 use strict;
  21         45  
  21         926  
8 21     21   141 use base 'MogileFS::Worker';
  21         50  
  21         3630  
9             use fields (
10 21         204 'fsck_queue_limit',
11             'repl_queue_limit',
12             'dele_queue_limit',
13             'rebl_queue_limit',
14 21     21   131 );
  21         49  
15 21     21   1564 use MogileFS::Util qw(every error debug encode_url_args);
  21         44  
  21         1688  
16 21     21   169 use MogileFS::Config;
  21         52  
  21         3048  
17 21     21   134 use MogileFS::Server;
  21         46  
  21         592  
18              
19 21     21   134 use constant DEF_FSCK_QUEUE_MAX => 20_000;
  21         59  
  21         1323  
20 21     21   116 use constant DEF_FSCK_QUEUE_INJECT => 1000;
  21         48  
  21         991  
21              
22 21     21   122 use constant DEF_REBAL_QUEUE_MAX => 10_000;
  21         62  
  21         1022  
23 21     21   135 use constant DEF_REBAL_QUEUE_INJECT => 500;
  21         42  
  21         46413  
24              
25             sub new {
26 0     0 0   my ($class, $psock) = @_;
27 0           my $self = fields::new($class);
28 0           $self->SUPER::new($psock);
29              
30 0           return $self;
31             }
32              
33 0     0 0   sub watchdog_timeout { 120; }
34              
35             # heartbeat all of the queues constantly.
36             # if a queue drops below a watermark, check for more work.
37             # NOTE: Uh. now that I think about it, should queue_check just return
38             # the status for all queues in one roundtrip? :(
39             # It's separate in case future workers want to manage their own queues, or
40             # this gets split up...
41             sub work {
42 0     0 0   my $self = shift;
43              
44 0           $self->{fsck_queue_limit} = 100;
45 0           $self->{repl_queue_limit} = 100;
46 0           $self->{dele_queue_limit} = 100;
47 0           $self->{rebl_queue_limit} = 100;
48              
49 0     0     Danga::Socket->AddOtherFds($self->psock_fd, sub{ $self->read_from_parent });
  0            
50              
51             # kick off the initial run
52 0           $self->check_queues;
53 0           Danga::Socket->EventLoop;
54             }
55              
56             # 'pings' parent and populates all queues.
57             sub check_queues {
58 0     0 0   my $self = shift;
59              
60 0           my $active = 0;
61 0 0         if ($self->validate_dbh) {
62 0           $self->send_to_parent("queue_depth all");
63 0           my $sto = Mgd::get_store();
64 0           $self->parent_ping;
65 0           $active += $self->_check_replicate_queues($sto);
66 0           $active += $self->_check_delete_queues($sto);
67 0           $active += $self->_check_fsck_queues($sto);
68 0           $active += $self->_check_rebal_queues($sto);
69             }
70              
71             # don't sleep if active (just avoid recursion)
72 0 0   0     Danga::Socket->AddTimer($active ? 0 : 1, sub { $self->check_queues });
  0            
73             }
74              
75             sub _check_delete_queues {
76 0     0     my $self = shift;
77 0           my $sto = shift;
78 0           my ($need_fetch, $new_limit) =
79             queue_depth_check($self->queue_depth('delete'),
80             $self->{dele_queue_limit});
81 0 0         return unless $need_fetch;
82 0           my @to_del = $sto->grab_files_to_delete2($new_limit);
83 0 0         $self->{dele_queue_limit} = @to_del ? $new_limit : 100;
84 0 0         return unless @to_del;
85 0           for my $todo (@to_del) {
86 0           $self->send_to_parent("queue_todo delete " .
87             encode_url_args($todo));
88             }
89 0           return 1;
90             }
91              
92             # NOTE: we only maintain one queue per worker, but we can easily
93             # give specialized work per worker by tagging the $todo href.
94             # in the case of replication, we want a normal "replication" queue,
95             # but also "drain" and "rebalance" queues. So use $todo->{type} or something.
96             # Drain/rebalance will be way awesomer with a queue attached:
97             # "drain 5% of devid 5" or "drain 10G off devids 7,8,9"
98             # hell, drain barely works if you encounter errors. Using a work queue
99             # should fix that.
100             # FIXME: Don't hardcode the min queue depth.
101             sub _check_replicate_queues {
102 0     0     my $self = shift;
103 0           my $sto = shift;
104 0           my ($need_fetch, $new_limit) =
105             queue_depth_check($self->queue_depth('replicate'),
106             $self->{repl_queue_limit});
107 0 0         return unless $need_fetch;
108 0           my @to_repl = $sto->grab_files_to_replicate($new_limit);
109 0 0         $self->{repl_queue_limit} = @to_repl ? $new_limit : 100;
110 0 0         return unless @to_repl;
111             # don't need to shuffle or sort, since we're the only tracker to get this
112             # list.
113 0           for my $todo (@to_repl) {
114 0           $todo->{_type} = 'replicate'; # could be 'drain', etc.
115 0           $self->send_to_parent("queue_todo replicate " .
116             encode_url_args($todo));
117             }
118 0           return 1;
119             }
120              
121             # FSCK is going to be a little odd... We still need a single "global"
122             # fsck worker to do the queue injection, but need to locally poll data.
123             sub _check_fsck_queues {
124 0     0     my $self = shift;
125 0           my $sto = shift;
126 0           my $fhost = MogileFS::Config->server_setting_cached('fsck_host');
127 0 0 0       if ($fhost && $fhost eq MogileFS::Config->hostname) {
128 0           $self->_inject_fsck_queues($sto);
129             }
130              
131             # Queue depth algorithm:
132             # if internal queue is less than 30% full, fetch more.
133             # if internal queue bottomed out, increase fetch limit by 50.
134             # fetch more work
135             # if no work fetched, reset limit to 100 (default)
136 0           my ($need_fetch, $new_limit) =
137             queue_depth_check($self->queue_depth('fsck'),
138             $self->{fsck_queue_limit});
139 0 0         return unless $need_fetch;
140 0           my @to_fsck = $sto->grab_files_to_queued(FSCK_QUEUE,
141             'type, flags', $new_limit);
142 0 0         $self->{fsck_queue_limit} = @to_fsck ? $new_limit : 100;
143 0 0         return unless @to_fsck;
144 0           for my $todo (@to_fsck) {
145 0           $self->send_to_parent("queue_todo fsck " . encode_url_args($todo));
146             }
147 0           return 1;
148             }
149              
150             sub _inject_fsck_queues {
151 0     0     my $self = shift;
152 0           my $sto = shift;
153              
154 0           $sto->fsck_log_summarize;
155 0           my $queue_size = $sto->file_queue_length(FSCK_QUEUE);
156 0   0       my $max_queue =
157             MogileFS::Config->server_setting_cached('queue_size_for_fsck') ||
158             DEF_FSCK_QUEUE_MAX;
159 0 0         return if ($queue_size >= $max_queue);
160              
161 0   0       my $max_checked = MogileFS::Config->server_setting('fsck_highest_fid_checked') || 0;
162 0           my $fid_at_end = MogileFS::Config->server_setting('fsck_fid_at_end');
163 0   0       my $to_inject =
164             MogileFS::Config->server_setting_cached('queue_rate_for_fsck') ||
165             DEF_FSCK_QUEUE_INJECT;
166 0           my $fids = $sto->get_fidids_between($max_checked, $fid_at_end, $to_inject);
167 0 0         unless (@$fids) {
168 0           MogileFS::Config->set_server_setting('fsck_highest_fid_checked',
169             $max_checked);
170              
171             # set these last since tests/scripts may rely on these to
172             # determine when fsck (injection) is complete
173 0           $sto->set_server_setting("fsck_host", undef);
174 0           $sto->set_server_setting("fsck_stop_time", $sto->get_db_unixtime);
175 0           return;
176             }
177              
178 0           $sto->enqueue_many_for_todo($fids, FSCK_QUEUE, 0);
179              
180 0           my $nmax = $fids->[-1];
181 0           MogileFS::Config->set_server_setting('fsck_highest_fid_checked', $nmax);
182             }
183              
184             sub _check_rebal_queues {
185 0     0     my $self = shift;
186 0           my $sto = shift;
187 0           my $rhost = MogileFS::Config->server_setting_cached('rebal_host');
188 0 0 0       if ($rhost && $rhost eq MogileFS::Config->hostname) {
189 0           $self->_inject_rebalance_queues($sto);
190             }
191              
192 0           my ($need_fetch, $new_limit) =
193             queue_depth_check($self->queue_depth('rebalance'),
194             $self->{rebl_queue_limit});
195 0 0         return unless $need_fetch;
196 0           my @to_rebal = $sto->grab_files_to_queued(REBAL_QUEUE,
197             'type, flags, devid, arg', $new_limit);
198 0 0         $self->{rebl_queue_limit} = @to_rebal ? $new_limit : 100;
199 0 0         return unless @to_rebal;
200 0           for my $todo (@to_rebal) {
201 0           $todo->{_type} = 'rebalance';
202 0           $self->send_to_parent("queue_todo rebalance " . encode_url_args($todo));
203             }
204 0           return 1;
205             }
206              
207             sub _inject_rebalance_queues {
208 0     0     my $self = shift;
209 0           my $sto = shift;
210              
211 0           my $queue_size = $sto->file_queue_length(REBAL_QUEUE);
212 0   0       my $max_queue =
213             MogileFS::Config->server_setting_cached('queue_size_for_rebal') ||
214             DEF_REBAL_QUEUE_MAX;
215 0 0         return if ($queue_size >= $max_queue);
216              
217 0   0       my $to_inject =
218             MogileFS::Config->server_setting_cached('queue_rate_for_rebal') ||
219             DEF_REBAL_QUEUE_INJECT;
220              
221             # TODO: Cache the rebal object. Requires explicitly blowing it up at the
222             # end of a run or ... I guess whenever the host sees it's not the rebal
223             # host.
224 0           my $rebal = MogileFS::Rebalance->new;
225 0           my $signal = MogileFS::Config->server_setting('rebal_signal');
226 0           my $rebal_pol = MogileFS::Config->server_setting('rebal_policy');
227 0           my $rebal_state = MogileFS::Config->server_setting('rebal_state');
228 0           $rebal->policy($rebal_pol);
229              
230 0           my @devs = Mgd::device_factory()->get_all;
231 0 0         if ($rebal_state) {
232 0           $rebal->load_state($rebal_state);
233             } else {
234 0           $rebal->init(\@devs);
235             }
236              
237             # Stopping is done via signal so we can note stop time in the state,
238             # and un-drain any devices that should be un-drained.
239 0 0 0       if ($signal && $signal eq 'stop') {
240 0           $rebal->stop;
241 0           $rebal_state = $rebal->save_state;
242 0           $sto->set_server_setting('rebal_signal', undef);
243 0           $sto->set_server_setting("rebal_host", undef);
244 0           $sto->set_server_setting('rebal_state', $rebal_state);
245 0           return;
246             }
247              
248 0           my $devfids = $rebal->next_fids_to_rebalance(\@devs, $sto, $to_inject);
249              
250             # undefined means there's no work left.
251 0 0         if (! defined $devfids) {
252             # Append some info to a rebalance log table?
253             # Leave state in the system for inspection post-run.
254             # TODO: Emit some sort of syslog/status line.
255 0           $rebal->finish;
256 0           $rebal_state = $rebal->save_state;
257 0           $sto->set_server_setting('rebal_state', $rebal_state);
258 0           $sto->set_server_setting("rebal_host", undef);
259 0           return;
260             }
261              
262             # Empty means nothing to queue this round.
263 0 0         if (@$devfids) {
264             # I wish there was less data serialization in the world.
265 0           map { $_->[2] = join(',', @{$_->[2]}) } @$devfids;
  0            
  0            
266 0           $sto->enqueue_many_for_todo($devfids, REBAL_QUEUE, 0);
267             }
268              
269 0           $rebal_state = $rebal->save_state;
270 0           MogileFS::Config->set_server_setting("rebal_state", $rebal_state);
271             }
272              
273             # takes the current queue depth and fetch limit
274             # returns whether or not to fetch, and new fetch limit.
275             # TODO: separate a fetch limit from a queue limit...
276             # so we don't hammer the DB with giant transactions, but loop
277             # fast trying to keep the queue full.
278             sub queue_depth_check {
279 0   0 0 0   my $max_limit =
280             MogileFS::Config->server_setting_cached('internal_queue_limit')
281             || 500;
282              
283 0           my ($depth, $limit) = @_;
284 0 0         if ($depth == 0) {
    0          
285 0 0         $limit += 50 unless $limit >= $max_limit;
286 0           return (1, $limit);
287             } elsif ($depth / $limit < 0.70) {
288 0           return (1, $limit);
289             }
290 0           return (0, $limit);
291             }
292              
293             1;
294              
295             # Local Variables:
296             # mode: perl
297             # c-basic-indent: 4
298             # indent-tabs-mode: nil
299             # End: