]> git.sur5r.net Git - bacula/bacula/blob - bacula/src/dird/ua_prune.c
Try to fix #1660 about segfault during pruning
[bacula/bacula] / bacula / src / dird / ua_prune.c
1 /*
2    Bacula® - The Network Backup Solution
3
4    Copyright (C) 2002-2009 Free Software Foundation Europe e.V.
5
6    The main author of Bacula is Kern Sibbald, with contributions from
7    many others, a complete list can be found in the file AUTHORS.
8    This program is Free Software; you can redistribute it and/or
9    modify it under the terms of version three of the GNU Affero General Public
10    License as published by the Free Software Foundation and included
11    in the file LICENSE.
12
13    This program is distributed in the hope that it will be useful, but
14    WITHOUT ANY WARRANTY; without even the implied warranty of
15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16    General Public License for more details.
17
18    You should have received a copy of the GNU Affero General Public License
19    along with this program; if not, write to the Free Software
20    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
21    02110-1301, USA.
22
23    Bacula® is a registered trademark of Kern Sibbald.
24    The licensor of Bacula is the Free Software Foundation Europe
25    (FSFE), Fiduciary Program, Sumatrastrasse 25, 8006 Zürich,
26    Switzerland, email:ftf@fsfeurope.org.
27 */
28 /*
29  *
30  *   Bacula Director -- User Agent Database prune Command
31  *      Applies retention periods
32  *
33  *     Kern Sibbald, February MMII
34  *
35  */
36
37 #include "bacula.h"
38 #include "dird.h"
39
40 /* Imported functions */
41
42 /* Forward referenced functions */
43 static bool grow_del_list(struct del_ctx *del);
44
45 /*
46  * Called here to count entries to be deleted
47  */
48 int del_count_handler(void *ctx, int num_fields, char **row)
49 {
50    struct s_count_ctx *cnt = (struct s_count_ctx *)ctx;
51
52    if (row[0]) {
53       cnt->count = str_to_int64(row[0]);
54    } else {
55       cnt->count = 0;
56    }
57    return 0;
58 }
59
60
61 /*
62  * Called here to make in memory list of JobIds to be
63  *  deleted and the associated PurgedFiles flag.
64  *  The in memory list will then be transversed
65  *  to issue the SQL DELETE commands.  Note, the list
66  *  is allowed to get to MAX_DEL_LIST_LEN to limit the
67  *  maximum malloc'ed memory.
68  */
69 int job_delete_handler(void *ctx, int num_fields, char **row)
70 {
71    struct del_ctx *del = (struct del_ctx *)ctx;
72
73    if (!grow_del_list(del)) {
74       return 1;
75    }
76    del->JobId[del->num_ids] = (JobId_t)str_to_int64(row[0]);
77    Dmsg2(60, "job_delete_handler row=%d val=%d\n", del->num_ids, del->JobId[del->num_ids]);
78    del->PurgedFiles[del->num_ids++] = (char)str_to_int64(row[1]);
79    return 0;
80 }
81
82 int file_delete_handler(void *ctx, int num_fields, char **row)
83 {
84    struct del_ctx *del = (struct del_ctx *)ctx;
85
86    if (!grow_del_list(del)) {
87       return 1;
88    }
89    del->JobId[del->num_ids++] = (JobId_t)str_to_int64(row[0]);
90 // Dmsg2(150, "row=%d val=%d\n", del->num_ids-1, del->JobId[del->num_ids-1]);
91    return 0;
92 }
93
94 /*
95  *   Prune records from database
96  *
97  *    prune files (from) client=xxx [pool=yyy]
98  *    prune jobs (from) client=xxx [pool=yyy]
99  *    prune volume=xxx
100  *    prune stats
101  */
102 int prunecmd(UAContext *ua, const char *cmd)
103 {
104    DIRRES *dir;
105    CLIENT *client;
106    POOL *pool;
107    POOL_DBR pr;
108    MEDIA_DBR mr;
109    utime_t retention;
110    int kw;
111
112    static const char *keywords[] = {
113       NT_("Files"),
114       NT_("Jobs"),
115       NT_("Volume"),
116       NT_("Stats"),
117       NULL};
118
119    if (!open_client_db(ua)) {
120       return false;
121    }
122
123    /* First search args */
124    kw = find_arg_keyword(ua, keywords);
125    if (kw < 0 || kw > 3) {
126       /* no args, so ask user */
127       kw = do_keyword_prompt(ua, _("Choose item to prune"), keywords);
128    }
129
130    switch (kw) {
131    case 0:  /* prune files */
132       client = get_client_resource(ua);
133       if (find_arg_with_value(ua, "pool") >= 0) {
134          pool = get_pool_resource(ua);
135       } else {
136          pool = NULL;
137       }
138       /* Pool File Retention takes precedence over client File Retention */
139       if (pool && pool->FileRetention > 0) {
140          if (!confirm_retention(ua, &pool->FileRetention, "File")) {
141             return false;
142          }
143       } else if (!client || !confirm_retention(ua, &client->FileRetention, "File")) {
144          return false;
145       }
146       prune_files(ua, client, pool);
147       return true;
148    case 1:  /* prune jobs */
149       client = get_client_resource(ua);
150       if (find_arg_with_value(ua, "pool") >= 0) {
151          pool = get_pool_resource(ua);
152       } else {
153          pool = NULL;
154       }
155       /* Pool Job Retention takes precedence over client Job Retention */
156       if (pool && pool->JobRetention > 0) {
157          if (!confirm_retention(ua, &pool->JobRetention, "Job")) {
158             return false;
159          }
160       } else if (!client || !confirm_retention(ua, &client->JobRetention, "Job")) {
161          return false;
162       }
163       /* ****FIXME**** allow user to select JobType */
164       prune_jobs(ua, client, pool, JT_BACKUP);
165       return 1;
166    case 2:  /* prune volume */
167       if (!select_pool_and_media_dbr(ua, &pr, &mr)) {
168          return false;
169       }
170       if (mr.Enabled == 2) {
171          ua->error_msg(_("Cannot prune Volume \"%s\" because it is archived.\n"),
172             mr.VolumeName);
173          return false;
174       }
175       if (!confirm_retention(ua, &mr.VolRetention, "Volume")) {
176          return false;
177       }
178       prune_volume(ua, &mr);
179       return true;
180    case 3:  /* prune stats */
181       dir = (DIRRES *)GetNextRes(R_DIRECTOR, NULL);
182       if (!dir->stats_retention) {
183          return false;
184       }
185       retention = dir->stats_retention;
186       if (!confirm_retention(ua, &retention, "Statistics")) {
187          return false;
188       }
189       prune_stats(ua, retention);
190       return true;
191    default:
192       break;
193    }
194
195    return true;
196 }
197
198 /* Prune Job stat records from the database. 
199  *
200  */
201 int prune_stats(UAContext *ua, utime_t retention)
202 {
203    char ed1[50];
204    POOL_MEM query(PM_MESSAGE);
205    utime_t now = (utime_t)time(NULL);
206
207    db_lock(ua->db);
208    Mmsg(query, "DELETE FROM JobHisto WHERE JobTDate < %s", 
209         edit_int64(now - retention, ed1));
210    db_sql_query(ua->db, query.c_str(), NULL, NULL);
211    db_unlock(ua->db);
212
213    ua->info_msg(_("Pruned Jobs from JobHisto catalog.\n"));
214
215    return true;
216 }
217
218 /*
219  * Prune File records from the database. For any Job which
220  * is older than the retention period, we unconditionally delete
221  * all File records for that Job.  This is simple enough that no
222  * temporary tables are needed. We simply make an in memory list of
223  * the JobIds meeting the prune conditions, then delete all File records
224  * pointing to each of those JobIds.
225  *
226  * This routine assumes you want the pruning to be done. All checking
227  *  must be done before calling this routine.
228  *
229  * Note: pool can possibly be NULL.
230  */
231 int prune_files(UAContext *ua, CLIENT *client, POOL *pool)
232 {
233    struct del_ctx del;
234    struct s_count_ctx cnt;
235    POOL_MEM query(PM_MESSAGE);
236    utime_t now, period;
237    CLIENT_DBR cr;
238    char ed1[50], ed2[50];
239
240    db_lock(ua->db);
241    memset(&cr, 0, sizeof(cr));
242    memset(&del, 0, sizeof(del));
243    bstrncpy(cr.Name, client->name(), sizeof(cr.Name));
244    if (!db_create_client_record(ua->jcr, ua->db, &cr)) {
245       db_unlock(ua->db);
246       return 0;
247    }
248
249    if (pool && pool->FileRetention > 0) {
250       period = pool->FileRetention;
251    } else {
252       period = client->FileRetention;
253    }
254    now = (utime_t)time(NULL);
255
256 //   edit_utime(now-period, ed1, sizeof(ed1));
257 //   Jmsg(ua->jcr, M_INFO, 0, _("Begin pruning Jobs older than %s secs.\n"), ed1);
258    Jmsg(ua->jcr, M_INFO, 0, _("Begin pruning Jobs.\n"));
259    /* Select Jobs -- for counting */ 
260    edit_int64(now - period, ed1);
261    Mmsg(query, count_select_job, ed1, edit_int64(cr.ClientId, ed2));
262    Dmsg3(050, "select now=%u period=%u sql=%s\n", (uint32_t)now, 
263                (uint32_t)period, query.c_str());
264    cnt.count = 0;
265    if (!db_sql_query(ua->db, query.c_str(), del_count_handler, (void *)&cnt)) {
266       ua->error_msg("%s", db_strerror(ua->db));
267       Dmsg0(050, "Count failed\n");
268       goto bail_out;
269    }
270
271    if (cnt.count == 0) {
272       if (ua->verbose) {
273          ua->warning_msg(_("No Files found to prune.\n"));
274       }
275       goto bail_out;
276    }
277
278    if (cnt.count < MAX_DEL_LIST_LEN) {
279       del.max_ids = cnt.count + 1;
280    } else {
281       del.max_ids = MAX_DEL_LIST_LEN;
282    }
283    del.tot_ids = 0;
284
285    del.JobId = (JobId_t *)malloc(sizeof(JobId_t) * del.max_ids);
286
287    /* Now process same set but making a delete list */
288    Mmsg(query, select_job, edit_int64(now - period, ed1), 
289         edit_int64(cr.ClientId, ed2));
290    db_sql_query(ua->db, query.c_str(), file_delete_handler, (void *)&del);
291
292    purge_files_from_job_list(ua, del);
293
294    edit_uint64_with_commas(del.num_del, ed1);
295    ua->info_msg(_("Pruned Files from %s Jobs for client %s from catalog.\n"),
296       ed1, client->name());
297
298 bail_out:
299    db_unlock(ua->db);
300    if (del.JobId) {
301       free(del.JobId);
302    }
303    return 1;
304 }
305
306
307 static void drop_temp_tables(UAContext *ua)
308 {
309    int i;
310    for (i=0; drop_deltabs[i]; i++) {
311       db_sql_query(ua->db, drop_deltabs[i], NULL, (void *)NULL);
312    }
313 }
314
315 static bool create_temp_tables(UAContext *ua)
316 {
317    /* Create temp tables and indicies */
318    if (!db_sql_query(ua->db, create_deltabs[db_type], NULL, (void *)NULL)) {
319       ua->error_msg("%s", db_strerror(ua->db));
320       Dmsg0(050, "create DelTables table failed\n");
321       return false;
322    }
323    if (!db_sql_query(ua->db, create_delindex, NULL, (void *)NULL)) {
324        ua->error_msg("%s", db_strerror(ua->db));
325        Dmsg0(050, "create DelInx1 index failed\n");
326        return false;
327    }
328    return true;
329 }
330
331 static bool grow_del_list(struct del_ctx *del)
332 {
333    if (del->num_ids == MAX_DEL_LIST_LEN) {
334       return false;
335    }
336
337    if (del->num_ids == del->max_ids) {
338       del->max_ids = (del->max_ids * 3) / 2;
339       del->JobId = (JobId_t *)brealloc(del->JobId, sizeof(JobId_t) *
340          del->max_ids);
341       del->PurgedFiles = (char *)brealloc(del->PurgedFiles, del->max_ids);
342    }
343    return true;
344 }
345
346 struct accurate_check_ctx {
347    DBId_t ClientId;                   /* Id of client */
348    DBId_t FileSetId;                  /* Id of FileSet */ 
349 };
350
351 /* row: Job.Name, FileSet, Client.Name, FileSetId, ClientId, Type */
352 static int job_select_handler(void *ctx, int num_fields, char **row)
353 {
354    alist *lst = (alist *)ctx;
355    struct accurate_check_ctx *res;
356    ASSERT(num_fields == 6);
357
358    /* If this job doesn't exist anymore in the configuration, delete it */
359    if (GetResWithName(R_JOB, row[0]) == NULL) {
360       return 0;
361    }
362
363    /* If this fileset doesn't exist anymore in the configuration, delete it */
364    if (GetResWithName(R_FILESET, row[1]) == NULL) {
365       return 0;
366    }
367
368    /* If this client doesn't exist anymore in the configuration, delete it */
369    if (GetResWithName(R_CLIENT, row[2]) == NULL) {
370       return 0;
371    }
372
373    /* Don't compute accurate things for Verify jobs */
374    if (*row[5] == 'V') {
375       return 0;
376    }
377
378    res = (struct accurate_check_ctx*) malloc(sizeof(struct accurate_check_ctx));
379    res->FileSetId = str_to_int64(row[3]);
380    res->ClientId = str_to_int64(row[4]);
381    lst->append(res);
382
383 // Dmsg2(150, "row=%d val=%d\n", del->num_ids-1, del->JobId[del->num_ids-1]);
384    return 0;
385 }
386
387 /*
388  * Pruning Jobs is a bit more complicated than purging Files
389  * because we delete Job records only if there is a more current
390  * backup of the FileSet. Otherwise, we keep the Job record.
391  * In other words, we never delete the only Job record that
392  * contains a current backup of a FileSet. This prevents the
393  * Volume from being recycled and destroying a current backup.
394  *
395  * For Verify Jobs, we do not delete the last InitCatalog.
396  *
397  * For Restore Jobs there are no restrictions.
398  */
399 int prune_jobs(UAContext *ua, CLIENT *client, POOL *pool, int JobType)
400 {
401    struct del_ctx del;
402    POOL_MEM query(PM_MESSAGE);
403    utime_t now, period;
404    CLIENT_DBR cr ;
405    char ed1[50], ed2[50];
406    alist *jobids_check=NULL;
407    struct accurate_check_ctx *elt;
408    db_list_ctx jobids, tempids;
409    JOB_DBR jr;
410
411    db_lock(ua->db);
412    memset(&del, 0, sizeof(del));
413    memset(&cr, 0, sizeof(cr));
414
415    bstrncpy(cr.Name, client->name(), sizeof(cr.Name));
416    if (!db_create_client_record(ua->jcr, ua->db, &cr)) {
417       db_unlock(ua->db);
418       return 0;
419    }
420
421    if (pool && pool->JobRetention > 0) {
422       period = pool->JobRetention;
423    } else {
424       period = client->JobRetention;
425    }
426    now = (utime_t)time(NULL);
427
428    /* Drop any previous temporary tables still there */
429    drop_temp_tables(ua);
430
431    /* Create temp tables and indicies */
432    if (!create_temp_tables(ua)) {
433       goto bail_out;
434    }
435
436    edit_utime(period, ed1, sizeof(ed1));
437    Jmsg(ua->jcr, M_INFO, 0, _("Begin pruning Jobs older than %s.\n"), ed1);
438
439    edit_int64(now - period, ed1); /* Jobs older than ed1 are good candidates */
440    edit_int64(cr.ClientId, ed2);
441
442    del.max_ids = 100;
443    del.JobId = (JobId_t *)malloc(sizeof(JobId_t) * del.max_ids);
444    del.PurgedFiles = (char *)malloc(del.max_ids);
445
446    /*
447     * Select all files that are older than the JobRetention period
448     *  and add them into the "DeletionCandidates" table.
449     */
450    Mmsg(query, 
451         "INSERT INTO DelCandidates "
452           "SELECT JobId,PurgedFiles,FileSetId,JobFiles,JobStatus "
453             "FROM Job "
454            "WHERE Type IN ('B', 'C', 'M', 'V',  'D', 'R', 'c', 'm', 'g') "
455              "AND JobTDate<%s AND ClientId=%s", 
456         ed1, ed2);
457
458    if (!db_sql_query(ua->db, query.c_str(), NULL, (void *)NULL)) {
459       if (ua->verbose) {
460          ua->error_msg("%s", db_strerror(ua->db));
461       }
462       goto bail_out;
463    }
464
465    /* Now, for the selection, we discard some of them in order to be always
466     * able to restore files. (ie, last full, last diff, last incrs)
467     * Note: The DISTINCT could be more useful if we don't get FileSetId
468     */
469    jobids_check = New(alist(10, owned_by_alist));
470    Mmsg(query, 
471 "SELECT DISTINCT Job.Name, FileSet, Client.Name, Job.FileSetId, "
472                 "Job.ClientId, Job.Type "
473   "FROM DelCandidates "
474        "JOIN Job USING (JobId) "
475        "JOIN Client USING (ClientId) "
476        "JOIN FileSet ON (Job.FileSetId = FileSet.FileSetId) "
477  "WHERE Job.Type IN ('B') "               /* Look only Backup jobs */
478    "AND Job.JobStatus IN ('T', 'W') "     /* Look only useful jobs */
479       );
480
481    /* The job_select_handler will skip jobs or filesets that are no longer
482     * in the configuration file. Interesting ClientId/FileSetId will be
483     * added to jobids_check
484     */
485    if (!db_sql_query(ua->db, query.c_str(), job_select_handler, jobids_check)) {
486       ua->error_msg("%s", db_strerror(ua->db));
487    }
488
489    /* For this selection, we exclude current jobs used for restore or
490     * accurate. This will prevent to prune the last full backup used for
491     * current backup & restore
492     */
493    memset(&jr, 0, sizeof(jr));
494    /* To find useful jobs, we do like an incremental */
495    jr.JobLevel = L_INCREMENTAL; 
496    foreach_alist(elt, jobids_check) {
497       jr.ClientId = elt->ClientId;   /* should be always the same */
498       jr.FileSetId = elt->FileSetId;
499       db_accurate_get_jobids(ua->jcr, ua->db, &jr, &tempids);
500       jobids.cat(tempids);
501    }
502
503    /* Discard latest Verify level=InitCatalog job 
504     * TODO: can have multiple fileset
505     */
506    Mmsg(query, 
507         "SELECT JobId, JobTDate "
508           "FROM Job "
509          "WHERE JobTDate<%s AND ClientId=%s "
510            "AND Type='V'    AND Level='V' "
511          "ORDER BY JobTDate DESC LIMIT 1", 
512         ed1, ed2);
513
514    if (!db_sql_query(ua->db, query.c_str(), db_list_handler, &jobids)) {
515       ua->error_msg("%s", db_strerror(ua->db));
516    }
517
518    /* If we found jobs to exclude from the DelCandidates list, we should
519     * also remove BaseJobs that can be linked with them
520     */
521    if (jobids.count > 0) {
522       Dmsg1(60, "jobids to exclude before basejobs = %s\n", jobids.list);
523       /* We also need to exclude all basejobs used */
524       db_get_used_base_jobids(ua->jcr, ua->db, jobids.list, &jobids);
525
526       /* Removing useful jobs from the DelCandidates list */
527       Mmsg(query, "DELETE FROM DelCandidates "
528                    "WHERE JobId IN (%s) "        /* JobId used in accurate */
529                      "AND JobFiles!=0",          /* Discard when JobFiles=0 */
530            jobids.list);
531
532       if (!db_sql_query(ua->db, query.c_str(), NULL, NULL)) {
533          ua->error_msg("%s", db_strerror(ua->db));
534          goto bail_out;         /* Don't continue if the list isn't clean */
535       }
536       Dmsg1(60, "jobids to exclude = %s\n", jobids.list);
537    }
538
539    /* We use DISTINCT because we can have two times the same job */
540    Mmsg(query, 
541         "SELECT DISTINCT DelCandidates.JobId,DelCandidates.PurgedFiles "
542           "FROM DelCandidates");
543    if (!db_sql_query(ua->db, query.c_str(), job_delete_handler, (void *)&del)) {
544       ua->error_msg("%s", db_strerror(ua->db));
545    }
546
547    purge_job_list_from_catalog(ua, del);
548
549    if (del.num_del > 0) {
550       ua->info_msg(_("Pruned %d %s for client %s from catalog.\n"), del.num_del,
551          del.num_del==1?_("Job"):_("Jobs"), client->name());
552     } else if (ua->verbose) {
553        ua->info_msg(_("No Jobs found to prune.\n"));
554     }
555
556 bail_out:
557    drop_temp_tables(ua);
558    db_unlock(ua->db);
559    if (del.JobId) {
560       free(del.JobId);
561    }
562    if (del.PurgedFiles) {
563       free(del.PurgedFiles);
564    }
565    if (jobids_check) {
566       delete jobids_check;
567    }
568    return 1;
569 }
570
571 /*
572  * Prune a given Volume
573  */
574 bool prune_volume(UAContext *ua, MEDIA_DBR *mr)
575 {
576    POOL_MEM query(PM_MESSAGE);
577    struct del_ctx del;
578    bool ok = false;
579    int count;
580
581    if (mr->Enabled == 2) {
582       return false;                   /* Cannot prune archived volumes */
583    }
584
585    memset(&del, 0, sizeof(del));
586    del.max_ids = 10000;
587    del.JobId = (JobId_t *)malloc(sizeof(JobId_t) * del.max_ids);
588
589    db_lock(ua->db);
590
591    /* Prune only Volumes with status "Full", or "Used" */
592    if (strcmp(mr->VolStatus, "Full")   == 0 ||
593        strcmp(mr->VolStatus, "Used")   == 0) {
594       Dmsg2(050, "get prune list MediaId=%d Volume %s\n", (int)mr->MediaId, mr->VolumeName);
595       count = get_prune_list_for_volume(ua, mr, &del);
596       Dmsg1(050, "Num pruned = %d\n", count);
597       if (count != 0) {
598          purge_job_list_from_catalog(ua, del);
599       }
600       ok = is_volume_purged(ua, mr);
601    }
602
603    db_unlock(ua->db);
604    if (del.JobId) {
605       free(del.JobId);
606    }
607    return ok;
608 }
609
610 /*
611  * Get prune list for a volume
612  */
613 int get_prune_list_for_volume(UAContext *ua, MEDIA_DBR *mr, del_ctx *del)
614 {
615    POOL_MEM query(PM_MESSAGE);
616    int count = 0;
617    utime_t now, period;
618    char ed1[50], ed2[50];
619
620    if (mr->Enabled == 2) {
621       return 0;                    /* cannot prune Archived volumes */
622    }
623
624    /*
625     * Now add to the  list of JobIds for Jobs written to this Volume
626     */
627    edit_int64(mr->MediaId, ed1); 
628    period = mr->VolRetention;
629    now = (utime_t)time(NULL);
630    edit_int64(now-period, ed2);
631    Mmsg(query, sel_JobMedia, ed1, ed2);
632    Dmsg3(250, "Now=%d period=%d now-period=%s\n", (int)now, (int)period,
633       ed2);
634
635    Dmsg1(050, "Query=%s\n", query.c_str());
636    if (!db_sql_query(ua->db, query.c_str(), file_delete_handler, (void *)del)) {
637       if (ua->verbose) {
638          ua->error_msg("%s", db_strerror(ua->db));
639       }
640       Dmsg0(050, "Count failed\n");
641       goto bail_out;
642    }
643    count = exclude_running_jobs_from_list(del);
644    
645 bail_out:
646    return count;
647 }
648
649 /*
650  * We have a list of jobs to prune or purge. If any of them is
651  *   currently running, we set its JobId to zero which effectively
652  *   excludes it.
653  *
654  * Returns the number of jobs that can be prunned or purged.
655  *
656  */
657 int exclude_running_jobs_from_list(del_ctx *prune_list)
658 {
659    int count = 0;
660    JCR *jcr;
661    bool skip;
662    int i;          
663
664    /* Do not prune any job currently running */
665    for (i=0; i < prune_list->num_ids; i++) {
666       skip = false;
667       foreach_jcr(jcr) {
668          if (jcr->JobId == prune_list->JobId[i]) {
669             Dmsg2(050, "skip running job JobId[%d]=%d\n", i, (int)prune_list->JobId[i]);
670             prune_list->JobId[i] = 0;
671             skip = true;
672             break;
673          }
674       }
675       endeach_jcr(jcr);
676       if (skip) {
677          continue;  /* don't increment count */
678       }
679       Dmsg2(050, "accept JobId[%d]=%d\n", i, (int)prune_list->JobId[i]);
680       count++;
681    }
682    return count;
683 }