lib/ntdb/ntdb.c

   1  /*
   2    Trivial Database 2: fetch, store and misc routines.
   3    Copyright (C) Rusty Russell 2010
   4
   5    This library is free software; you can redistribute it and/or
   6    modify it under the terms of the GNU Lesser General Public
   7    License as published by the Free Software Foundation; either
   8    version 3 of the License, or (at your option) any later version.
   9
  10    This library is distributed in the hope that it will be useful,
  11    but WITHOUT ANY WARRANTY; without even the implied warranty of
  12    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13    Lesser General Public License for more details.
  14
  15    You should have received a copy of the GNU Lesser General Public
  16    License along with this library; if not, see <http://www.gnu.org/licenses/>.
  17 */
  18 #include "private.h"
  19 #ifndef HAVE_LIBREPLACE
  20 #include <ccan/asprintf/asprintf.h>
  21 #include <stdarg.h>
  22 #endif
  23
  24 static enum NTDB_ERROR update_rec_hdr(struct ntdb_context *ntdb,
  25                                      ntdb_off_t off,
  26                                      ntdb_len_t keylen,
  27                                      ntdb_len_t datalen,
  28                                      struct ntdb_used_record *rec,
  29                                      uint64_t h)
  30 {
  31         uint64_t dataroom = rec_data_length(rec) + rec_extra_padding(rec);
  32         enum NTDB_ERROR ecode;
  33
  34         ecode = set_header(ntdb, rec, NTDB_USED_MAGIC, keylen, datalen,
  35                            keylen + dataroom, h);
  36         if (ecode == NTDB_SUCCESS) {
  37                 ecode = ntdb_write_convert(ntdb, off, rec, sizeof(*rec));
  38         }
  39         return ecode;
  40 }
  41
  42 static enum NTDB_ERROR replace_data(struct ntdb_context *ntdb,
  43                                    struct hash_info *h,
  44                                    NTDB_DATA key, NTDB_DATA dbuf,
  45                                    ntdb_off_t old_off, ntdb_len_t old_room,
  46                                    bool growing)
  47 {
  48         ntdb_off_t new_off;
  49         enum NTDB_ERROR ecode;
  50
  51         /* Allocate a new record. */
  52         new_off = alloc(ntdb, key.dsize, dbuf.dsize, h->h, NTDB_USED_MAGIC,
  53                         growing);
  54         if (NTDB_OFF_IS_ERR(new_off)) {
  55                 return NTDB_OFF_TO_ERR(new_off);
  56         }
  57
  58         /* We didn't like the existing one: remove it. */
  59         if (old_off) {
  60                 ntdb->stats.frees++;
  61                 ecode = add_free_record(ntdb, old_off,
  62                                         sizeof(struct ntdb_used_record)
  63                                         + key.dsize + old_room,
  64                                         NTDB_LOCK_WAIT, true);
  65                 if (ecode == NTDB_SUCCESS)
  66                         ecode = replace_in_hash(ntdb, h, new_off);
  67         } else {
  68                 ecode = add_to_hash(ntdb, h, new_off);
  69         }
  70         if (ecode != NTDB_SUCCESS) {
  71                 return ecode;
  72         }
  73
  74         new_off += sizeof(struct ntdb_used_record);
  75         ecode = ntdb->io->twrite(ntdb, new_off, key.dptr, key.dsize);
  76         if (ecode != NTDB_SUCCESS) {
  77                 return ecode;
  78         }
  79
  80         new_off += key.dsize;
  81         ecode = ntdb->io->twrite(ntdb, new_off, dbuf.dptr, dbuf.dsize);
  82         if (ecode != NTDB_SUCCESS) {
  83                 return ecode;
  84         }
  85
  86         if (ntdb->flags & NTDB_SEQNUM)
  87                 ntdb_inc_seqnum(ntdb);
  88
  89         return NTDB_SUCCESS;
  90 }
  91
  92 static enum NTDB_ERROR update_data(struct ntdb_context *ntdb,
  93                                   ntdb_off_t off,
  94                                   NTDB_DATA dbuf,
  95                                   ntdb_len_t extra)
  96 {
  97         enum NTDB_ERROR ecode;
  98
  99         ecode = ntdb->io->twrite(ntdb, off, dbuf.dptr, dbuf.dsize);
 100         if (ecode == NTDB_SUCCESS && extra) {
 101                 /* Put a zero in; future versions may append other data. */
 102                 ecode = ntdb->io->twrite(ntdb, off + dbuf.dsize, "", 1);
 103         }
 104         if (ntdb->flags & NTDB_SEQNUM)
 105                 ntdb_inc_seqnum(ntdb);
 106
 107         return ecode;
 108 }
 109
 110 _PUBLIC_ enum NTDB_ERROR ntdb_store(struct ntdb_context *ntdb,
 111                          NTDB_DATA key, NTDB_DATA dbuf, int flag)
 112 {
 113         struct hash_info h;
 114         ntdb_off_t off;
 115         ntdb_len_t old_room = 0;
 116         struct ntdb_used_record rec;
 117         enum NTDB_ERROR ecode;
 118
 119         off = find_and_lock(ntdb, key, F_WRLCK, &h, &rec, NULL);
 120         if (NTDB_OFF_IS_ERR(off)) {
 121                 return NTDB_OFF_TO_ERR(off);
 122         }
 123
 124         /* Now we have lock on this hash bucket. */
 125         if (flag == NTDB_INSERT) {
 126                 if (off) {
 127                         ecode = NTDB_ERR_EXISTS;
 128                         goto out;
 129                 }
 130         } else {
 131                 if (off) {
 132                         old_room = rec_data_length(&rec)
 133                                 + rec_extra_padding(&rec);
 134                         if (old_room >= dbuf.dsize) {
 135                                 /* Can modify in-place.  Easy! */
 136                                 ecode = update_rec_hdr(ntdb, off,
 137                                                        key.dsize, dbuf.dsize,
 138                                                        &rec, h.h);
 139                                 if (ecode != NTDB_SUCCESS) {
 140                                         goto out;
 141                                 }
 142                                 ecode = update_data(ntdb,
 143                                                     off + sizeof(rec)
 144                                                     + key.dsize, dbuf,
 145                                                     old_room - dbuf.dsize);
 146                                 if (ecode != NTDB_SUCCESS) {
 147                                         goto out;
 148                                 }
 149                                 ntdb_unlock_hashes(ntdb, h.hlock_start,
 150                                                   h.hlock_range, F_WRLCK);
 151                                 return NTDB_SUCCESS;
 152                         }
 153                 } else {
 154                         if (flag == NTDB_MODIFY) {
 155                                 /* if the record doesn't exist and we
 156                                    are in NTDB_MODIFY mode then we should fail
 157                                    the store */
 158                                 ecode = NTDB_ERR_NOEXIST;
 159                                 goto out;
 160                         }
 161                 }
 162         }
 163
 164         /* If we didn't use the old record, this implies we're growing. */
 165         ecode = replace_data(ntdb, &h, key, dbuf, off, old_room, off);
 166 out:
 167         ntdb_unlock_hashes(ntdb, h.hlock_start, h.hlock_range, F_WRLCK);
 168         return ecode;
 169 }
 170
 171 _PUBLIC_ enum NTDB_ERROR ntdb_append(struct ntdb_context *ntdb,
 172                           NTDB_DATA key, NTDB_DATA dbuf)
 173 {
 174         struct hash_info h;
 175         ntdb_off_t off;
 176         struct ntdb_used_record rec;
 177         ntdb_len_t old_room = 0, old_dlen;
 178         unsigned char *newdata;
 179         NTDB_DATA new_dbuf;
 180         enum NTDB_ERROR ecode;
 181
 182         off = find_and_lock(ntdb, key, F_WRLCK, &h, &rec, NULL);
 183         if (NTDB_OFF_IS_ERR(off)) {
 184                 return NTDB_OFF_TO_ERR(off);
 185         }
 186
 187         if (off) {
 188                 old_dlen = rec_data_length(&rec);
 189                 old_room = old_dlen + rec_extra_padding(&rec);
 190
 191                 /* Fast path: can append in place. */
 192                 if (rec_extra_padding(&rec) >= dbuf.dsize) {
 193                         ecode = update_rec_hdr(ntdb, off, key.dsize,
 194                                                old_dlen + dbuf.dsize, &rec,
 195                                                h.h);
 196                         if (ecode != NTDB_SUCCESS) {
 197                                 goto out;
 198                         }
 199
 200                         off += sizeof(rec) + key.dsize + old_dlen;
 201                         ecode = update_data(ntdb, off, dbuf,
 202                                             rec_extra_padding(&rec));
 203                         goto out;
 204                 }
 205
 206                 /* Slow path. */
 207                 newdata = ntdb->alloc_fn(ntdb, key.dsize + old_dlen + dbuf.dsize,
 208                                      ntdb->alloc_data);
 209                 if (!newdata) {
 210                         ecode = ntdb_logerr(ntdb, NTDB_ERR_OOM, NTDB_LOG_ERROR,
 211                                            "ntdb_append:"
 212                                            " failed to allocate %zu bytes",
 213                                            (size_t)(key.dsize + old_dlen
 214                                                     + dbuf.dsize));
 215                         goto out;
 216                 }
 217                 ecode = ntdb->io->tread(ntdb, off + sizeof(rec) + key.dsize,
 218                                        newdata, old_dlen);
 219                 if (ecode != NTDB_SUCCESS) {
 220                         goto out_free_newdata;
 221                 }
 222                 memcpy(newdata + old_dlen, dbuf.dptr, dbuf.dsize);
 223                 new_dbuf.dptr = newdata;
 224                 new_dbuf.dsize = old_dlen + dbuf.dsize;
 225         } else {
 226                 newdata = NULL;
 227                 new_dbuf = dbuf;
 228         }
 229
 230         /* If they're using ntdb_append(), it implies they're growing record. */
 231         ecode = replace_data(ntdb, &h, key, new_dbuf, off, old_room, true);
 232
 233 out_free_newdata:
 234         ntdb->free_fn(newdata, ntdb->alloc_data);
 235 out:
 236         ntdb_unlock_hashes(ntdb, h.hlock_start, h.hlock_range, F_WRLCK);
 237         return ecode;
 238 }
 239
 240 _PUBLIC_ enum NTDB_ERROR ntdb_fetch(struct ntdb_context *ntdb, NTDB_DATA key,
 241                          NTDB_DATA *data)
 242 {
 243         ntdb_off_t off;
 244         struct ntdb_used_record rec;
 245         struct hash_info h;
 246         enum NTDB_ERROR ecode;
 247
 248         off = find_and_lock(ntdb, key, F_RDLCK, &h, &rec, NULL);
 249         if (NTDB_OFF_IS_ERR(off)) {
 250                 return NTDB_OFF_TO_ERR(off);
 251         }
 252
 253         if (!off) {
 254                 ecode = NTDB_ERR_NOEXIST;
 255         } else {
 256                 data->dsize = rec_data_length(&rec);
 257                 data->dptr = ntdb_alloc_read(ntdb, off + sizeof(rec) + key.dsize,
 258                                             data->dsize);
 259                 if (NTDB_PTR_IS_ERR(data->dptr)) {
 260                         ecode = NTDB_PTR_ERR(data->dptr);
 261                 } else
 262                         ecode = NTDB_SUCCESS;
 263         }
 264
 265         ntdb_unlock_hashes(ntdb, h.hlock_start, h.hlock_range, F_RDLCK);
 266         return ecode;
 267 }
 268
 269 _PUBLIC_ bool ntdb_exists(struct ntdb_context *ntdb, NTDB_DATA key)
 270 {
 271         ntdb_off_t off;
 272         struct ntdb_used_record rec;
 273         struct hash_info h;
 274
 275         off = find_and_lock(ntdb, key, F_RDLCK, &h, &rec, NULL);
 276         if (NTDB_OFF_IS_ERR(off)) {
 277                 return false;
 278         }
 279         ntdb_unlock_hashes(ntdb, h.hlock_start, h.hlock_range, F_RDLCK);
 280
 281         return off ? true : false;
 282 }
 283
 284 _PUBLIC_ enum NTDB_ERROR ntdb_delete(struct ntdb_context *ntdb, NTDB_DATA key)
 285 {
 286         ntdb_off_t off;
 287         struct ntdb_used_record rec;
 288         struct hash_info h;
 289         enum NTDB_ERROR ecode;
 290
 291         off = find_and_lock(ntdb, key, F_WRLCK, &h, &rec, NULL);
 292         if (NTDB_OFF_IS_ERR(off)) {
 293                 return NTDB_OFF_TO_ERR(off);
 294         }
 295
 296         if (!off) {
 297                 ecode = NTDB_ERR_NOEXIST;
 298                 goto unlock;
 299         }
 300
 301         ecode = delete_from_hash(ntdb, &h);
 302         if (ecode != NTDB_SUCCESS) {
 303                 goto unlock;
 304         }
 305
 306         /* Free the deleted entry. */
 307         ntdb->stats.frees++;
 308         ecode = add_free_record(ntdb, off,
 309                                 sizeof(struct ntdb_used_record)
 310                                 + rec_key_length(&rec)
 311                                 + rec_data_length(&rec)
 312                                 + rec_extra_padding(&rec),
 313                                 NTDB_LOCK_WAIT, true);
 314
 315         if (ntdb->flags & NTDB_SEQNUM)
 316                 ntdb_inc_seqnum(ntdb);
 317
 318 unlock:
 319         ntdb_unlock_hashes(ntdb, h.hlock_start, h.hlock_range, F_WRLCK);
 320         return ecode;
 321 }
 322
 323 _PUBLIC_ unsigned int ntdb_get_flags(struct ntdb_context *ntdb)
 324 {
 325         return ntdb->flags;
 326 }
 327
 328 static bool inside_transaction(const struct ntdb_context *ntdb)
 329 {
 330         return ntdb->transaction != NULL;
 331 }
 332
 333 static bool readonly_changable(struct ntdb_context *ntdb, const char *caller)
 334 {
 335         if (inside_transaction(ntdb)) {
 336                 ntdb_logerr(ntdb, NTDB_ERR_EINVAL, NTDB_LOG_USE_ERROR,
 337                             "%s: can't change"
 338                             " NTDB_RDONLY inside transaction",
 339                             caller);
 340                 return false;
 341         }
 342         return true;
 343 }
 344
 345 _PUBLIC_ void ntdb_add_flag(struct ntdb_context *ntdb, unsigned flag)
 346 {
 347         if (ntdb->flags & NTDB_INTERNAL) {
 348                 ntdb_logerr(ntdb, NTDB_ERR_EINVAL, NTDB_LOG_USE_ERROR,
 349                             "ntdb_add_flag: internal db");
 350                 return;
 351         }
 352         switch (flag) {
 353         case NTDB_NOLOCK:
 354                 ntdb->flags |= NTDB_NOLOCK;
 355                 break;
 356         case NTDB_NOMMAP:
 357                 ntdb->flags |= NTDB_NOMMAP;
 358 #ifndef HAVE_INCOHERENT_MMAP
 359                 ntdb_munmap(ntdb->file);
 360 #endif
 361                 break;
 362         case NTDB_NOSYNC:
 363                 ntdb->flags |= NTDB_NOSYNC;
 364                 break;
 365         case NTDB_SEQNUM:
 366                 ntdb->flags |= NTDB_SEQNUM;
 367                 break;
 368         case NTDB_ALLOW_NESTING:
 369                 ntdb->flags |= NTDB_ALLOW_NESTING;
 370                 break;
 371         case NTDB_RDONLY:
 372                 if (readonly_changable(ntdb, "ntdb_add_flag"))
 373                         ntdb->flags |= NTDB_RDONLY;
 374                 break;
 375         default:
 376                 ntdb_logerr(ntdb, NTDB_ERR_EINVAL, NTDB_LOG_USE_ERROR,
 377                             "ntdb_add_flag: Unknown flag %u", flag);
 378         }
 379 }
 380
 381 _PUBLIC_ void ntdb_remove_flag(struct ntdb_context *ntdb, unsigned flag)
 382 {
 383         if (ntdb->flags & NTDB_INTERNAL) {
 384                 ntdb_logerr(ntdb, NTDB_ERR_EINVAL, NTDB_LOG_USE_ERROR,
 385                             "ntdb_remove_flag: internal db");
 386                 return;
 387         }
 388         switch (flag) {
 389         case NTDB_NOLOCK:
 390                 ntdb->flags &= ~NTDB_NOLOCK;
 391                 break;
 392         case NTDB_NOMMAP:
 393                 ntdb->flags &= ~NTDB_NOMMAP;
 394 #ifndef HAVE_INCOHERENT_MMAP
 395                 /* If mmap incoherent, we were mmaping anyway. */
 396                 ntdb_mmap(ntdb);
 397 #endif
 398                 break;
 399         case NTDB_NOSYNC:
 400                 ntdb->flags &= ~NTDB_NOSYNC;
 401                 break;
 402         case NTDB_SEQNUM:
 403                 ntdb->flags &= ~NTDB_SEQNUM;
 404                 break;
 405         case NTDB_ALLOW_NESTING:
 406                 ntdb->flags &= ~NTDB_ALLOW_NESTING;
 407                 break;
 408         case NTDB_RDONLY:
 409                 if ((ntdb->open_flags & O_ACCMODE) == O_RDONLY) {
 410                         ntdb_logerr(ntdb, NTDB_ERR_EINVAL, NTDB_LOG_USE_ERROR,
 411                                     "ntdb_remove_flag: can't"
 412                                     " remove NTDB_RDONLY on ntdb"
 413                                     " opened with O_RDONLY");
 414                         break;
 415                 }
 416                 if (readonly_changable(ntdb, "ntdb_remove_flag"))
 417                         ntdb->flags &= ~NTDB_RDONLY;
 418                 break;
 419         default:
 420                 ntdb_logerr(ntdb, NTDB_ERR_EINVAL, NTDB_LOG_USE_ERROR,
 421                             "ntdb_remove_flag: Unknown flag %u",
 422                             flag);
 423         }
 424 }
 425
 426 _PUBLIC_ const char *ntdb_errorstr(enum NTDB_ERROR ecode)
 427 {
 428         /* Gcc warns if you miss a case in the switch, so use that. */
 429         switch (NTDB_ERR_TO_OFF(ecode)) {
 430         case NTDB_ERR_TO_OFF(NTDB_SUCCESS): return "Success";
 431         case NTDB_ERR_TO_OFF(NTDB_ERR_CORRUPT): return "Corrupt database";
 432         case NTDB_ERR_TO_OFF(NTDB_ERR_IO): return "IO Error";
 433         case NTDB_ERR_TO_OFF(NTDB_ERR_LOCK): return "Locking error";
 434         case NTDB_ERR_TO_OFF(NTDB_ERR_OOM): return "Out of memory";
 435         case NTDB_ERR_TO_OFF(NTDB_ERR_EXISTS): return "Record exists";
 436         case NTDB_ERR_TO_OFF(NTDB_ERR_EINVAL): return "Invalid parameter";
 437         case NTDB_ERR_TO_OFF(NTDB_ERR_NOEXIST): return "Record does not exist";
 438         case NTDB_ERR_TO_OFF(NTDB_ERR_RDONLY): return "write not permitted";
 439         }
 440         return "Invalid error code";
 441 }
 442
 443 enum NTDB_ERROR COLD ntdb_logerr(struct ntdb_context *ntdb,
 444                                enum NTDB_ERROR ecode,
 445                                enum ntdb_log_level level,
 446                                const char *fmt, ...)
 447 {
 448         char *message;
 449         va_list ap;
 450         size_t len;
 451         /* ntdb_open paths care about errno, so save it. */
 452         int saved_errno = errno;
 453
 454         if (!ntdb->log_fn)
 455                 return ecode;
 456
 457         va_start(ap, fmt);
 458         len = vsnprintf(NULL, 0, fmt, ap);
 459         va_end(ap);
 460
 461         message = ntdb->alloc_fn(ntdb, len + 1, ntdb->alloc_data);
 462         if (!message) {
 463                 ntdb->log_fn(ntdb, NTDB_LOG_ERROR, NTDB_ERR_OOM,
 464                             "out of memory formatting message:", ntdb->log_data);
 465                 ntdb->log_fn(ntdb, level, ecode, fmt, ntdb->log_data);
 466         } else {
 467                 va_start(ap, fmt);
 468                 vsnprintf(message, len+1, fmt, ap);
 469                 va_end(ap);
 470                 ntdb->log_fn(ntdb, level, ecode, message, ntdb->log_data);
 471                 ntdb->free_fn(message, ntdb->alloc_data);
 472         }
 473         errno = saved_errno;
 474         return ecode;
 475 }
 476
 477 _PUBLIC_ enum NTDB_ERROR ntdb_parse_record_(struct ntdb_context *ntdb,
 478                                  NTDB_DATA key,
 479                                  enum NTDB_ERROR (*parse)(NTDB_DATA k,
 480                                                          NTDB_DATA d,
 481                                                          void *data),
 482                                  void *data)
 483 {
 484         ntdb_off_t off;
 485         struct ntdb_used_record rec;
 486         struct hash_info h;
 487         enum NTDB_ERROR ecode;
 488
 489         off = find_and_lock(ntdb, key, F_RDLCK, &h, &rec, NULL);
 490         if (NTDB_OFF_IS_ERR(off)) {
 491                 return NTDB_OFF_TO_ERR(off);
 492         }
 493
 494         if (!off) {
 495                 ecode = NTDB_ERR_NOEXIST;
 496         } else {
 497                 const void *dptr;
 498                 dptr = ntdb_access_read(ntdb, off + sizeof(rec) + key.dsize,
 499                                        rec_data_length(&rec), false);
 500                 if (NTDB_PTR_IS_ERR(dptr)) {
 501                         ecode = NTDB_PTR_ERR(dptr);
 502                 } else {
 503                         NTDB_DATA d = ntdb_mkdata(dptr, rec_data_length(&rec));
 504
 505                         ecode = parse(key, d, data);
 506                         ntdb_access_release(ntdb, dptr);
 507                 }
 508         }
 509
 510         ntdb_unlock_hashes(ntdb, h.hlock_start, h.hlock_range, F_RDLCK);
 511         return ecode;
 512 }
 513
 514 _PUBLIC_ const char *ntdb_name(const struct ntdb_context *ntdb)
 515 {
 516         return ntdb->name;
 517 }
 518
 519 _PUBLIC_ int64_t ntdb_get_seqnum(struct ntdb_context *ntdb)
 520 {
 521         return ntdb_read_off(ntdb, offsetof(struct ntdb_header, seqnum));
 522 }
 523
 524
 525 _PUBLIC_ int ntdb_fd(const struct ntdb_context *ntdb)
 526 {
 527         return ntdb->file->fd;
 528 }
 529
 530 struct traverse_state {
 531         enum NTDB_ERROR error;
 532         struct ntdb_context *dest_db;
 533 };
 534
 535 /*
 536   traverse function for repacking
 537  */
 538 static int repack_traverse(struct ntdb_context *ntdb, NTDB_DATA key, NTDB_DATA data,
 539                            struct traverse_state *state)
 540 {
 541         state->error = ntdb_store(state->dest_db, key, data, NTDB_INSERT);
 542         if (state->error != NTDB_SUCCESS) {
 543                 return -1;
 544         }
 545         return 0;
 546 }
 547
 548 _PUBLIC_ enum NTDB_ERROR ntdb_repack(struct ntdb_context *ntdb)
 549 {
 550         struct ntdb_context *tmp_db;
 551         struct traverse_state state;
 552
 553         state.error = ntdb_transaction_start(ntdb);
 554         if (state.error != NTDB_SUCCESS) {
 555                 return state.error;
 556         }
 557
 558         tmp_db = ntdb_open("tmpdb", NTDB_INTERNAL, O_RDWR|O_CREAT, 0, NULL);
 559         if (tmp_db == NULL) {
 560                 state.error = ntdb_logerr(ntdb, NTDB_ERR_OOM, NTDB_LOG_ERROR,
 561                                          __location__
 562                                          " Failed to create tmp_db");
 563                 ntdb_transaction_cancel(ntdb);
 564                 return state.error;
 565         }
 566
 567         state.dest_db = tmp_db;
 568         if (ntdb_traverse(ntdb, repack_traverse, &state) < 0) {
 569                 goto fail;
 570         }
 571
 572         state.error = ntdb_wipe_all(ntdb);
 573         if (state.error != NTDB_SUCCESS) {
 574                 goto fail;
 575         }
 576
 577         state.dest_db = ntdb;
 578         if (ntdb_traverse(tmp_db, repack_traverse, &state) < 0) {
 579                 goto fail;
 580         }
 581
 582         ntdb_close(tmp_db);
 583         return ntdb_transaction_commit(ntdb);
 584
 585 fail:
 586         ntdb_transaction_cancel(ntdb);
 587         ntdb_close(tmp_db);
 588         return state.error;
 589 }