source/exsinglesv/filelock.c

0001 /**
0002  * @brief File locking routines
0003  *
0004  * @file filelock.c
0005  */
0006 /* -----------------------------------------------------------------------------
0007  * Enduro/X Middleware Platform for Distributed Transaction Processing
0008  * Copyright (C) 2009-2016, ATR Baltic, Ltd. All Rights Reserved.
0009  * Copyright (C) 2017-2023, Mavimax, Ltd. All Rights Reserved.
0010  * This software is released under one of the following licenses:
0011  * AGPL (with Java and Go exceptions) or Mavimax's license for commercial use.
0012  * See LICENSE file for full text.
0013  * -----------------------------------------------------------------------------
0014  * AGPL license:
0015  *
0016  * This program is free software; you can redistribute it and/or modify it under
0017  * the terms of the GNU Affero General Public License, version 3 as published
0018  * by the Free Software Foundation;
0019  *
0020  * This program is distributed in the hope that it will be useful, but WITHOUT ANY
0021  * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
0022  * PARTICULAR PURPOSE. See the GNU Affero General Public License, version 3
0023  * for more details.
0024  *
0025  * You should have received a copy of the GNU Affero General Public License along
0026  * with this program; if not, write to the Free Software Foundation, Inc.,
0027  * 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
0028  *
0029  * -----------------------------------------------------------------------------
0030  * A commercial use license is available from Mavimax, Ltd
0031  * contact@mavimax.com
0032  * -----------------------------------------------------------------------------
0033  */
0034 #include <stdio.h>
0035 #include <stdlib.h>
0036 #include <string.h>
0037 #include <errno.h>
0038 #include <regex.h>
0039 #include <utlist.h>
0040 #include <unistd.h>
0041 #include <signal.h>
0042
0043 #include <ndebug.h>
0044 #include <atmi.h>
0045 #include <atmi_int.h>
0046 #include <typed_buf.h>
0047 #include <ndrstandard.h>
0048 #include <ubf.h>
0049 #include <ubfutil.h>
0050 #include <cconfig.h>
0051 #include "exsinglesv.h"
0052 #include <fcntl.h>
0053 #include <singlegrp.h>
0054 #include <Exfields.h>
0055 #include <nstdutil.h>
0056 /*---------------------------Externs------------------------------------*/
0057 /*---------------------------Macros-------------------------------------*/
0058 #define MAX_LOCKS       2
0059 #define DATA_SIZE       24 /* full block for node */
0060 #define DATA_FOR_CRC    16 /* full block for node */
0061 #define PING_READ_CRC_ERR  -2
0062 #define PING_NO_FILE  -3
0063 /*---------------------------Enums--------------------------------------*/
0064 /*---------------------------Typedefs-----------------------------------*/
0065
0066 /**
0067  * Locking structure
0068 */
0069 typedef struct
0070 {
0071     int fd;         /**< file-fd locked         */
0072     char lockfile[PATH_MAX+1]; /**< lock file              */
0073 } ndrx_exsinglesv_filelock_t;
0074
0075 /*---------------------------Globals------------------------------------*/
0076 /*---------------------------Statics------------------------------------*/
0077 /**
0078  * Locks used.
0079  */
0080 exprivate ndrx_exsinglesv_filelock_t M_locks[MAX_LOCKS] = {
0081         {.fd=EXFAIL, .lockfile=""},
0082         {.fd=EXFAIL, .lockfile=""}
0083 };
0084
0085 /*---------------------------Prototypes---------------------------------*/
0086
0087 /**
0088  * Check PID of the the lock file (file must be locked)
0089  * @param lock_no lock number (file numbers)
0090  * @param lockfile lock file
0091  * @return EXSUCCEED/EXFAIL
0092  */
0093 expublic int ndrx_exsinglesv_file_chkpid(int lock_no, char *lockfile)
0094 {
0095     int ret = EXSUCCEED;
0096     struct flock lck;
0097
0098     memset(&lck, 0, sizeof(lck));
0099
0100     lck.l_whence = SEEK_SET;
0101     lck.l_start = 0;
0102     lck.l_len = 1;
0103     lck.l_type = F_WRLCK;
0104
0105     TP_LOG(log_debug, "Checking (%d) [%s] lock status...", lock_no, lockfile);
0106
0107     if ( EXSUCCEED!=fcntl(M_locks[lock_no].fd, F_GETLK, &lck))
0108     {
0109         TP_LOG(log_info, "Failed to fcntl F_GETLK on [%s]: %s",
0110                     lockfile, strerror(errno));
0111         EXFAIL_OUT(ret);
0112     }
0113
0114     /*
0115      * we as lock owners, can lock it twice, thus must be reported as F_UNLOCK
0116      */
0117     if (lck.l_type!=F_UNLCK)
0118     {
0119         TP_LOG(log_error, "ERROR ! Lock file [%s] locked by other process (pid=%d)",
0120                     lockfile, (int)lck.l_pid);
0121         EXFAIL_OUT(ret);
0122     }
0123
0124 out:
0125     return ret;
0126 }
0127
0128 /**
0129  * Perform locking on the file
0130  * @param lock_no lock number
0131  * @param lockfile lock file
0132  * @return EXSUCCEED/EXFAIL
0133  */
0134 expublic int ndrx_exsinglesv_file_lock(int lock_no, char *lockfile)
0135 {
0136     int ret = EXSUCCEED;
0137     struct flock lck;
0138
0139     memset(&lck, 0, sizeof(lck));
0140
0141     lck.l_whence = SEEK_SET;
0142     lck.l_start = 0;
0143     lck.l_len = 1;
0144     lck.l_type = F_WRLCK;
0145
0146     TP_LOG(log_debug, "Trying to lock file (%d) [%s]", lock_no, lockfile);
0147
0148     /* open lock file */
0149     if (EXFAIL==(M_locks[lock_no].fd = open(lockfile, O_RDWR|O_CREAT, 0666)))
0150     {
0151         TP_LOG(log_error, "Failed to open lock file [%s]: %s",
0152                 lockfile, strerror(errno));
0153         ret=NDRX_LOCKE_FILE;
0154         goto out;
0155     }
0156
0157     /* lock the region with fcntl() */
0158     if (EXFAIL==fcntl(M_locks[lock_no].fd, F_SETLK, &lck))
0159     {
0160         if (EACCES==errno || EAGAIN==errno)
0161         {
0162             /* memset(&lck, 0, sizeof(lck)); */
0163             fcntl(M_locks[lock_no].fd, F_GETLK, &lck);
0164
0165             TP_LOG(log_info, "Failed to lock file [%s]: %s (pid=%d)",
0166                     lockfile, strerror(errno), (int)lck.l_pid);
0167             ret=NDRX_LOCKE_BUSY;
0168             goto out;
0169         }
0170         else
0171         {
0172             TP_LOG(log_error, "Failed to lock file [%s]: %s",
0173                 lockfile, strerror(errno));
0174             ret=NDRX_LOCKE_LOCK;
0175             goto out;
0176         }
0177     }
0178
0179     NDRX_STRCPY_SAFE(M_locks[lock_no].lockfile, lockfile);
0180
0181 out:
0182
0183     /* close the file in case of lock failure */
0184     if (EXSUCCEED!=ret && M_locks[lock_no].fd!=EXFAIL)
0185     {
0186         close(M_locks[lock_no].fd);
0187         M_locks[lock_no].fd = EXFAIL;
0188     }
0189     return ret;
0190 }
0191
0192 /**
0193  * Unlock the file
0194  * @param lock_no lock number
0195  * @return EXSUCCEED/EXFAIL
0196  */
0197 expublic int ndrx_exsinglesv_file_unlock(int lock_no)
0198 {
0199     int ret = EXSUCCEED;
0200     struct flock lock;
0201
0202     if (EXFAIL==M_locks[lock_no].fd)
0203     {
0204         TP_LOG(log_error, "Lock file [%s] is not locked",
0205                 M_locks[lock_no].lockfile);
0206         ret=NDRX_LOCKE_NOLOCK;
0207         goto out;
0208     }
0209
0210     lock.l_type = F_UNLCK;  /* Unlock */
0211     lock.l_whence = SEEK_SET;
0212     lock.l_start = 0;
0213     lock.l_len = 1;  /* Unlock the entire file */
0214
0215     if (fcntl(M_locks[lock_no].fd, F_SETLK, &lock) == -1) {
0216         TP_LOG(log_error, "Failed to unlock [%s]: %s",
0217                 M_locks[lock_no].lockfile, strerror(errno));
0218         ret=NDRX_LOCKE_LOCK;
0219         goto out;
0220     }
0221
0222 out:
0223
0224     /* close anyway... */
0225     close(M_locks[lock_no].fd);
0226     M_locks[lock_no].fd = EXFAIL;
0227
0228     return ret;
0229 }
0230
0231 /**
0232  * Extended group check.
0233  * @param lock_ctx lock context
0234  * @param chk_files check files (do not call remote at all)
0235  * @return EXFAIL/EXTRUE/EXFALSE
0236  */
0237 expublic int ndrx_exsinglesv_sg_is_locked(ndrx_locksm_ctx_t *lock_ctx, int force_chk)
0238 {
0239     int ret=EXFALSE;
0240     UBFH *p_ub = NULL;
0241     long tmp, rsplen;
0242     char svcnm[MAXTIDENT+1]={EXEOS};
0243     int locked_by_stat;
0244     char lckstatus;
0245     long last_refresh;
0246     long their_sequence;
0247     int is_net_rply;
0248     int i;
0249     ndrx_exsinglesv_lockent_t ent;
0250
0251     /* check local */
0252     lock_ctx->pshm=ndrx_sg_get(ndrx_G_exsinglesv_conf.procgrp_lp_no);
0253     if (NULL==lock_ctx->pshm)
0254     {
0255         TP_LOG(log_error, "Failed to get singleton process group: %d",
0256                 ndrx_G_exsinglesv_conf.procgrp_lp_no);
0257         EXFAIL_OUT(ret);
0258     }
0259
0260     /* Load group locally... */
0261     ndrx_sg_load(&lock_ctx->local, lock_ctx->pshm);
0262
0263     ret = ndrx_sg_is_locked(ndrx_G_exsinglesv_conf.procgrp_lp_no, NULL, 0);
0264
0265     /* check all remote */
0266     if (EXTRUE!=ret)
0267     {
0268         TP_LOG(log_info, "Group %d is not locked by shm",
0269                 ndrx_G_exsinglesv_conf.procgrp_lp_no);
0270         goto out;
0271     }
0272
0273     if (!(lock_ctx->local.flags & NDRX_SG_VERIFY) && !force_chk)
0274     {
0275         TP_LOG(log_info, "Group %d verification is not required",
0276                 ndrx_G_exsinglesv_conf.procgrp_lp_no);
0277         goto out;
0278     }
0279
0280     /* do the checks of the nodes */
0281     for (i=0; i<CONF_NDRX_NODEID_COUNT; i++)
0282     {
0283         int locked_by_stat=EXFALSE;
0284         last_refresh = 0;
0285
0286         if (lock_ctx->local.sg_nodes[i])
0287         {
0288             if (lock_ctx->local.sg_nodes[i]==G_atmi_env.our_nodeid)
0289             {
0290                 continue;
0291             }
0292             TP_LOG(log_debug, "Checking node [%d]...",
0293                     lock_ctx->local.sg_nodes[i]);
0294
0295             if (NULL!=p_ub)
0296             {
0297                 tpfree((char *)p_ub);
0298                 p_ub = NULL;
0299             }
0300
0301             /* Try remote (if not disabled) */
0302             ret=EXSUCCEED;
0303             if (!ndrx_G_exsinglesv_conf.noremote)
0304             {
0305                 p_ub = (UBFH *)tpalloc("UBF", NULL, 1024);
0306
0307                 if (NULL==p_ub)
0308                 {
0309                     TP_LOG(log_error, "Failed to allocate UBF");
0310                     EXFAIL_OUT(ret);
0311                 }
0312
0313                 /* load query buffer */
0314
0315                 tmp = ndrx_G_exsinglesv_conf.procgrp_lp_no;
0316                 if (EXSUCCEED!=Bchg(p_ub, EX_COMMAND, 0, NDRX_SGCMD_QUERY, 0L)
0317                     || EXSUCCEED!=Bchg(p_ub, EX_PROCGRP_NO, 0, (char *)&tmp, 0L))
0318                 {
0319                     TP_LOG(log_error, "Failed to setup request buffer: %s",
0320                         Bstrerror(Berror));
0321                     EXFAIL_OUT(ret);
0322                 }
0323
0324                 tpsblktime(ndrx_G_exsinglesv_conf.svc_timeout, TPBLK_NEXT);
0325
0326                 /* call server for results */
0327                 snprintf(svcnm, sizeof(svcnm), NDRX_SVC_SGREM, (long)lock_ctx->local.sg_nodes[i],
0328                     ndrx_G_exsinglesv_conf.procgrp_lp_no);
0329                 TP_LOG(log_debug, "Checking with [%s] group %d lock status", svcnm,
0330                     ndrx_G_exsinglesv_conf.procgrp_lp_no);
0331
0332                 ret = tpcall(svcnm, (char*)p_ub, 0L, (char **)&p_ub, &rsplen, TPNOTRAN);
0333             }
0334
0335             /*
0336              * if have their lock status & their are locked.
0337              * we have lost the lock, return error.
0338              * (unlock the shm). We will reply, however
0339              * ndrxd may kill the group earlier.
0340              */
0341
0342             /*
0343              * if they are not locked from the service reply,
0344              * our lock time must be newer than theirs.
0345              * if not, we have lost the lock and return the error.
0346              * (unlock the shm)
0347              */
0348             if (EXSUCCEED!=ret || ndrx_G_exsinglesv_conf.noremote)
0349             {
0350                 int log_lev = log_error;
0351
0352                 /* previous was service call... */
0353                 if (EXSUCCEED!=ret)
0354                 {
0355                     if (TPESVCFAIL==tperrno)
0356                     {
0357                         /* dump the reply buffer... */
0358                         ndrx_tplogprintubf(log_error, "svc error response", p_ub);
0359                     }
0360                     else  if (TPENOENT==tperrno)
0361                     {
0362                         log_lev = log_debug;
0363                     }
0364
0365                     TP_LOG(log_lev, "Failed to call [%s]: %s - falling back to disk check",
0366                         svcnm, tpstrerror(tperrno));
0367
0368                     /*  succeed anyway */
0369                     ret=EXSUCCEED;
0370                 }
0371
0372                 /* read the node entry from the disk */
0373                 if (EXSUCCEED!=ndrx_exsinglesv_ping_read(lock_ctx->local.sg_nodes[i], &ent))
0374                 {
0375                     TP_LOG(log_error, "Failed to ping lock file [%s] for node %d"
0376                         , ndrx_G_exsinglesv_conf.lockfile_2, (int)lock_ctx->local.sg_nodes[i]);
0377                     userlog("Failed to ping lock file [%s] for node %d"
0378                         , ndrx_G_exsinglesv_conf.lockfile_2, (int)lock_ctx->local.sg_nodes[i]);
0379                     EXFAIL_OUT(ret);
0380                 }
0381
0382                 /* load the values down.. */
0383                 last_refresh = ent.lock_time;
0384                 their_sequence = ent.sequence;
0385                 is_net_rply=EXFALSE;
0386             }
0387             else
0388             {
0389                 ndrx_tplogprintubf(log_debug, "svc response", p_ub);
0390                 /* read lock status... */
0391                 if (EXSUCCEED!=Bget(p_ub, EX_LCKSTATUS, 0, &lckstatus, 0L)
0392                     || EXSUCCEED!=Bget(p_ub, EX_TSTAMP, 0, (char *)&last_refresh, 0L)
0393                     || EXSUCCEED!=Bget(p_ub, EX_SEQUENCE, 0, (char *)&their_sequence, 0L))
0394                 {
0395                     /* mandatory field is missing in reply */
0396                     TP_LOG(log_error, "Failed to get EX_LCKSTATUS/EX_TSTAMP/EX_SEQUENCE from node %d: %s",
0397                         (int)lock_ctx->local.sg_nodes[i], Bstrerror(Berror));
0398                     EXFAIL_OUT(ret);
0399                 }
0400
0401                 if (EXTRUE==lckstatus)
0402                 {
0403                     locked_by_stat=EXTRUE;
0404                 }
0405
0406                 is_net_rply=EXTRUE;
0407             }
0408
0409             /* take the decsion */
0410             if (locked_by_stat)
0411             {
0412                 TP_LOG(log_error, "Node %d is locked according to network status. "
0413                     "Their refresh time: %ld, our time: %ld - UNLOCKING local group %d",
0414                     (int)lock_ctx->local.sg_nodes[i], last_refresh, lock_ctx->local.last_refresh,
0415                     ndrx_G_exsinglesv_conf.procgrp_lp_no);
0416                 userlog("Node %d is locked according to network status. "
0417                     "Their refresh time: %ld, our time: %ld - UNLOCKING local group %d",
0418                     (int)lock_ctx->local.sg_nodes[i], last_refresh, lock_ctx->local.last_refresh,
0419                     ndrx_G_exsinglesv_conf.procgrp_lp_no);
0420
0421                 ndrx_sg_unlock(lock_ctx->pshm, NDRX_SG_RSN_NETLOCK);
0422                 EXFAIL_OUT(ret);
0423             }
0424
0425             if (their_sequence>=lock_ctx->local.sequence)
0426             {
0427                 TP_LOG(log_error, "Node %d has higher sequence than us. "
0428                     "their seq: %ld, our seq: %ld, their refresh time: %ld, our time: %ld - UNLOCKING local group %d. Source: %s",
0429                     (int)lock_ctx->local.sg_nodes[i], their_sequence, lock_ctx->local.sequence, last_refresh, lock_ctx->local.last_refresh,
0430                     ndrx_G_exsinglesv_conf.procgrp_lp_no, is_net_rply?"network":"disk");
0431                 userlog("Node %d has higher sequence than us. "
0432                     "their seq: %ld, our seq: %ld, their refresh time: %ld, our time: %ld - UNLOCKING local group %d. Source: %s",
0433                     (int)lock_ctx->local.sg_nodes[i], their_sequence, lock_ctx->local.sequence, last_refresh, lock_ctx->local.last_refresh,
0434                     ndrx_G_exsinglesv_conf.procgrp_lp_no, is_net_rply?"network":"disk");
0435
0436                 if (is_net_rply)
0437                 {
0438                     ndrx_sg_unlock(lock_ctx->pshm, NDRX_SG_RSN_NETSEQ);
0439                 }
0440                 else
0441                 {
0442                     ndrx_sg_unlock(lock_ctx->pshm, NDRX_SG_RSN_FSEQ);
0443                 }
0444
0445                 EXFAIL_OUT(ret);
0446             }
0447         }
0448         else
0449         {
0450             /* array is sorted out of linked nodes */
0451             break;
0452         }
0453     }
0454
0455     /* if we are here, then we are locked fine */
0456     ret=EXTRUE;
0457
0458     /* if any fail, check local */
0459 out:
0460
0461     if (NULL!=p_ub)
0462     {
0463         tpfree((char *)p_ub);
0464     }
0465
0466     return ret;
0467 }
0468
0469 /**
0470  * read 16 bytes for each of the nodes in the file
0471  * 8 bytes=> lock time, 8 bytes => crc32 (to match the E/X lib).
0472  * Note that no parallel locks are allowed. If any lock
0473  * is found on the file, it shall be reported as error.
0474  */
0475 expublic int ndrx_exsinglesv_ping_do(ndrx_locksm_ctx_t *lock_ctx)
0476 {
0477     int fd=EXFAIL;
0478     struct stat st;
0479     int ret=EXSUCCEED;
0480     ssize_t bytes_written;
0481     struct flock lock;
0482     size_t fsize = DATA_SIZE*CONF_NDRX_NODEID_COUNT*2;
0483     ndrx_exsinglesv_lockent_t ent;
0484     int i, copy;
0485     char data_block[DATA_SIZE];
0486
0487     /* write our statistics to two places ...
0488      * so that if we crash during the first write, the
0489      * last record sayes OK... and other node can recover from it
0490      * to keep the counter growing past the crash.
0491      */
0492     for (copy=0; copy<2; copy++)
0493     {
0494         size_t offset= DATA_SIZE * CONF_NDRX_NODEID_COUNT * copy
0495             + DATA_SIZE*(G_atmi_env.our_nodeid-1);
0496
0497         fd = open(ndrx_G_exsinglesv_conf.lockfile_2, O_RDWR | O_CREAT, 0660);
0498
0499         if (EXFAIL==fd)
0500         {
0501             TP_LOG(log_error, "copy %d: Failed to open [%s]: %s",
0502                 copy, ndrx_G_exsinglesv_conf.lockfile_2, strerror(errno));
0503             EXFAIL_OUT(ret);
0504         }
0505
0506         /* Acquire an fcntl() write lock */
0507         lock.l_type = F_WRLCK;
0508         lock.l_whence = SEEK_SET;
0509         lock.l_start = 0;
0510
0511         /* Lock the entire file */
0512         lock.l_len = 0;
0513
0514         if (fcntl(fd, F_SETLKW, &lock) == -1)
0515         {
0516             TP_LOG(log_error, "copy %d: Failed to write-lock [%s] (fd=%d) file: %s",
0517                 copy, ndrx_G_exsinglesv_conf.lockfile_2, fd, strerror(errno));
0518             EXFAIL_OUT(ret);
0519         }
0520
0521         if (fstat(fd, &st) == -1)
0522         {
0523             TP_LOG(log_error, "copy %d: Failed to stat [%s] (fd=%d) file: %s",
0524                 copy, ndrx_G_exsinglesv_conf.lockfile_2, fd, strerror(errno));
0525             EXFAIL_OUT(ret);
0526         }
0527
0528         /* initialize the lock file.. */
0529         if (st.st_size == 0)
0530         {
0531             if (ftruncate(fd, fsize) == -1)
0532             {
0533                 TP_LOG(log_error, "copy %d: Truncate [%s] (fd=%d) to %d bytes failed: %s",
0534                     copy, ndrx_G_exsinglesv_conf.lockfile_2, fd, fsize, strerror(errno));
0535                 EXFAIL_OUT(ret);
0536             }
0537
0538             /* reset structs to CRC32 valid */
0539             memset(data_block, 0, DATA_FOR_CRC);
0540             ent.crc32=ndrx_Crc32_ComputeBuf(0, (unsigned char *)data_block, DATA_FOR_CRC);
0541             ent.crc32 = htonll(ent.crc32);
0542             memcpy(data_block+DATA_FOR_CRC, &ent.crc32, sizeof(ent.crc32));
0543
0544             /* create location for secondary location too... */
0545             for (i=0; i<CONF_NDRX_NODEID_COUNT*2; i++)
0546             {
0547                 size_t init_offset=DATA_SIZE * i;
0548
0549                 if (lseek(fd, init_offset, SEEK_SET) == -1)
0550                 {
0551                     TP_LOG(log_error, "copy %d: lseek [%s] (fd=%d) to offset %d bytes failed: %s",
0552                             copy, ndrx_G_exsinglesv_conf.lockfile_2, fd, init_offset, strerror(errno));
0553                     EXFAIL_OUT(ret);
0554                 }
0555
0556                 bytes_written = write(fd, data_block, DATA_SIZE);
0557
0558                 /* shall succeed for such small amount... */
0559                 if (DATA_SIZE!=bytes_written)
0560                 {
0561                     TP_LOG(log_error, "copy %d, write [%s] (fd=%d) of %d bytes (wrote %zd) failed: %s",
0562                             copy, ndrx_G_exsinglesv_conf.lockfile_2, fd, DATA_SIZE,
0563                             bytes_written, strerror(errno));
0564                     EXFAIL_OUT(ret);
0565                 }
0566                 else
0567                 {
0568                     TP_LOG(log_debug, "copy %d: Wrote %zd bytes at offset %d", copy, bytes_written, offset);
0569                 }
0570             }
0571         }
0572
0573         if (lseek(fd, offset, SEEK_SET) == -1)
0574         {
0575             TP_LOG(log_error, "copy %d: lseek [%s] (fd=%d) to offset %d bytes failed: %s",
0576                     copy, ndrx_G_exsinglesv_conf.lockfile_2, fd, offset, strerror(errno));
0577             EXFAIL_OUT(ret);
0578         }
0579
0580         /* write new sequence down there... */
0581         /* and write new refresh time down there too... */
0582         ent.lock_time = htonll(lock_ctx->new_refresh);
0583         ent.sequence = htonll(lock_ctx->new_sequence);
0584
0585         /* copy to memory buffer */
0586         memcpy(data_block, &ent.lock_time, sizeof(ent.lock_time));
0587         memcpy(data_block+sizeof(ent.lock_time), &ent.sequence, sizeof(ent.sequence));
0588         ent.crc32=ndrx_Crc32_ComputeBuf(0, (unsigned char *)data_block, DATA_FOR_CRC);
0589
0590         /* Dump data... */
0591         TP_LOG(log_info, "copy %d: writting nodeid=%d group=%d "
0592             "refresh=%ld sequence=%ld crc32=%lx",
0593             copy, G_atmi_env.our_nodeid, ndrx_G_exsinglesv_conf.procgrp_lp_no,
0594             (long)lock_ctx->new_refresh, (long)lock_ctx->new_sequence, (long)ent.crc32);
0595
0596         ent.crc32 = htonll(ent.crc32);
0597         memcpy(data_block+DATA_FOR_CRC, &ent.crc32, sizeof(ent.crc32));
0598
0599         bytes_written = write(fd, data_block, DATA_SIZE);
0600
0601         /* shall succeed for such small amount... */
0602         if (DATA_SIZE!=bytes_written)
0603         {
0604             TP_LOG(log_error, "copy %d: write [%s] (fd=%d) of %d bytes (wrote %zd) failed: %s",
0605                     copy, ndrx_G_exsinglesv_conf.lockfile_2, fd, DATA_SIZE,
0606                     bytes_written, strerror(errno));
0607             EXFAIL_OUT(ret);
0608         }
0609         else
0610         {
0611             TP_LOG(log_debug, "copy %d: Wrote %zd bytes at offset %d",
0612                 copy, bytes_written, offset);
0613         }
0614
0615         /* flush the changes to the disk fully */
0616         if (EXSUCCEED!=fsync(fd))
0617         {
0618             TP_LOG(log_error, "copy %d: fsync [%s] (fd=%d) failed: %s",
0619                 copy, ndrx_G_exsinglesv_conf.lockfile_2, fd, strerror(errno));
0620             EXFAIL_OUT(ret);
0621         }
0622
0623         /* Release the lock and close the file */
0624         lock.l_type = F_UNLCK;
0625         if (fcntl(fd, F_SETLK, &lock) == -1)
0626         {
0627             TP_LOG(log_error, "copy %d: fcntl unlock [%s] (fd=%d) failed: %s",
0628                 copy, ndrx_G_exsinglesv_conf.lockfile_2, fd, strerror(errno));
0629             EXFAIL_OUT(ret);
0630         }
0631
0632         if (EXFAIL!=fd)
0633         {
0634             close(fd);
0635             fd=EXFAIL;
0636         }
0637     }
0638
0639 out:
0640
0641     if (EXFAIL!=fd)
0642     {
0643         close(fd);
0644     }
0645
0646     return ret;
0647 }
0648
0649 /**
0650  * Read the data from the ping file.
0651  *  This totally does 3x attempts:
0652  *   1) if first one failed, try to read from the second file
0653  *   2) if second failed, try first, just in case if first was in progress of write / crashed
0654  *   3) if second try again first, just in case if first was in progress of write
0655  * @param copy which copy to read
0656  * @param nodeid
0657  * @param p_ent entry to read
0658  * @return EXFAIL or UTC epoch lock time, or -2 if CRC error
0659  */
0660 exprivate int ndrx_exsinglesv_ping_read_int(int copy, int nodeid, ndrx_exsinglesv_lockent_t *p_ent)
0661 {
0662     int ret = EXSUCCEED;
0663     int fd;
0664     struct flock lock;
0665     ssize_t bytes_read;
0666     size_t offset=DATA_SIZE * CONF_NDRX_NODEID_COUNT * copy + DATA_SIZE*(nodeid-1);
0667     char data_block[DATA_SIZE];
0668     uint64_t crc32_calc;
0669
0670     /* Open the file for reading */
0671     fd = open(ndrx_G_exsinglesv_conf.lockfile_2, O_RDONLY);
0672     if (EXFAIL==fd)
0673     {
0674         TP_LOG(log_error, "Failed to open [%s]: %s",
0675             ndrx_G_exsinglesv_conf.lockfile_2, strerror(errno));
0676
0677         /* if file is not found, return PING_NO_FILE */
0678         if (ENOENT==errno)
0679         {
0680             ret=PING_NO_FILE;
0681             goto out;
0682         }
0683
0684         EXFAIL_OUT(ret);
0685     }
0686
0687     lock.l_type = F_RDLCK;
0688     lock.l_whence = SEEK_SET;
0689     lock.l_start = 0;
0690     lock.l_len = 0;
0691
0692     TP_LOG(log_debug, "Acquring read lock on fd %d", fd);
0693
0694     if (fcntl(fd, F_SETLKW, &lock) == EXFAIL)
0695     {
0696         TP_LOG(log_error, "copy %d: Failed to read lock [%s] (fd=%d) file: %s",
0697             copy, ndrx_G_exsinglesv_conf.lockfile_2, fd, strerror(errno));
0698         EXFAIL_OUT(ret);
0699     }
0700
0701     TP_LOG(log_debug, "Read locked on fd %d", fd);
0702
0703     if (lseek(fd, offset, SEEK_SET) == -1)
0704     {
0705         TP_LOG(log_error, "copy %d: lseek [%s] (fd=%d) to offset %d bytes failed: %s",
0706                 copy, ndrx_G_exsinglesv_conf.lockfile_2, fd, offset, strerror(errno));
0707         EXFAIL_OUT(ret);
0708     }
0709
0710     bytes_read = read(fd, data_block, DATA_SIZE);
0711     if (bytes_read != DATA_SIZE)
0712     {
0713         TP_LOG(log_error, "copy %d: read [%s] (fd=%d) of %d bytes (wrote %zd) failed: %s",
0714                 copy, ndrx_G_exsinglesv_conf.lockfile_2, fd, DATA_SIZE,
0715                 bytes_read, strerror(errno));
0716         EXFAIL_OUT(ret);
0717     }
0718     else
0719     {
0720         TP_LOG(log_debug, "copy %d: Read %zd bytes at offset %d: for node: %d (fd=%d)",
0721             copy, bytes_read, offset, nodeid, fd);
0722     }
0723
0724     /* Release the lock and close the file */
0725     lock.l_type = F_UNLCK;
0726     if (fcntl(fd, F_SETLK, &lock) == -1)
0727     {
0728         TP_LOG(log_error, "copy %d: fcntl unlock [%s] (fd=%d) failed: %s",
0729             copy, ndrx_G_exsinglesv_conf.lockfile_2, fd, strerror(errno));
0730         EXFAIL_OUT(ret);
0731     }
0732
0733     /* extract from buffer */
0734     memcpy(&p_ent->lock_time, data_block, sizeof(p_ent->lock_time));
0735     memcpy(&p_ent->sequence, data_block+sizeof(p_ent->lock_time), sizeof(p_ent->sequence));
0736     memcpy(&p_ent->crc32, data_block+DATA_FOR_CRC, sizeof(p_ent->crc32));
0737
0738     /* convert to host format */
0739     p_ent->lock_time = ntohll(p_ent->lock_time);
0740     p_ent->crc32 = ntohll(p_ent->crc32);
0741     p_ent->sequence = ntohll(p_ent->sequence);
0742
0743     TP_LOG(log_info, "copy %d: got nodeid=%d group=%d "
0744             "refresh=%lu sequence=%lu crc32=%lx",
0745             copy, nodeid, ndrx_G_exsinglesv_conf.procgrp_lp_no,
0746             (long)p_ent->lock_time, (long)p_ent->sequence, (long)p_ent->crc32);
0747
0748     crc32_calc = ndrx_Crc32_ComputeBuf(0, (unsigned char *)data_block, DATA_FOR_CRC);
0749
0750     if (p_ent->crc32!=crc32_calc)
0751     {
0752         TP_LOG(log_warn, "copy %d: CRC32 mismatch (disk: %lx calc: %lx) file [%s] offset: %d", copy,
0753             (long)p_ent->crc32, (long)crc32_calc, ndrx_G_exsinglesv_conf.lockfile_2, (int)offset);
0754         ret=PING_READ_CRC_ERR;
0755     }
0756
0757 out:
0758     if (EXFAIL!=fd)
0759     {
0760         close(fd);
0761     }
0762
0763     return ret;
0764 }
0765
0766 /**
0767  * Try to extract lock entry from file.
0768  * @param nodeid node id
0769  * @param p_ent entry to read
0770  * @return EXFAIL or EXSUCCEED (p_ent loaded with CRC32 validate values)
0771  */
0772 expublic int ndrx_exsinglesv_ping_read(int nodeid, ndrx_exsinglesv_lockent_t *p_ent)
0773 {
0774     int i, ret;
0775
0776     for (i=0; i<3; i++)
0777     {
0778         ret=ndrx_exsinglesv_ping_read_int(i<2?i:0, nodeid, p_ent);
0779
0780         if (ret==PING_READ_CRC_ERR)
0781         {
0782             continue;
0783         }
0784         else
0785         {
0786             /* we have a verified result */
0787             break;
0788         }
0789     }
0790
0791     /* give some advice if lock file is totally corrupt... */
0792     if (PING_READ_CRC_ERR==ret)
0793     {
0794         TP_LOG(log_error, "Failed to read ping lock file for node %d - corrupted file. "
0795             "Shutdown the cluster, remove lock file [%s] and boot again.",
0796             nodeid, ndrx_G_exsinglesv_conf.lockfile_2);
0797
0798         userlog("Failed to read ping lock file for node %d - corrupted file. "
0799             "Shutdown the cluster, remove lock file [%s] and boot again.",
0800             nodeid, ndrx_G_exsinglesv_conf.lockfile_2);
0801         EXFAIL_OUT(ret);
0802     }
0803
0804 out:
0805
0806     TP_LOG(log_info, "%s returns %d", __FUNCTION__, ret);
0807     return ret;
0808 }
0809
0810 /**
0811  * Read max sequence number from files for all the nodes...
0812  * Used during intial lock and pinging...
0813  * If file does not exist, j
0814  * @return max sequence number as seen in lock files
0815  */
0816 expublic long ndrx_exsinglesv_sg_max_seq(ndrx_locksm_ctx_t *lock_ctx)
0817 {
0818     int ret=EXSUCCEED;
0819     long seq_max=0;
0820     int i;
0821     ndrx_exsinglesv_lockent_t ent;
0822
0823     for (i=0; i<CONF_NDRX_NODEID_COUNT; i++)
0824     {
0825         if (lock_ctx->local.sg_nodes[i])
0826         {
0827             TP_LOG(log_debug, "Checking node [%d]...",
0828                     lock_ctx->local.sg_nodes[i]);
0829
0830             ret = ndrx_exsinglesv_ping_read(lock_ctx->local.sg_nodes[i], &ent);
0831
0832             /* no file, max sequence is 0 */
0833             if (ret==PING_NO_FILE)
0834             {
0835                 /* we are done... */
0836                 ret=0;
0837                 break;
0838             }
0839             else if (EXSUCCEED!=ret)
0840             {
0841                 TP_LOG(log_error, "Failed to distinguish current max sequence number for node %d",
0842                     (int)lock_ctx->local.sg_nodes[i]);
0843                 EXFAIL_OUT(ret);
0844             }
0845
0846             if (ent.sequence > seq_max)
0847             {
0848                 seq_max = ent.sequence;
0849             }
0850         }
0851     }
0852
0853 out:
0854     TP_LOG(log_info, "returns %d, max sequence %ld", ret, seq_max);
0855
0856     if (EXSUCCEED==ret)
0857     {
0858         return seq_max;
0859     }
0860     else
0861     {
0862         return ret;
0863     }
0864 }
0865
0866 /* vim: set ts=4 sw=4 et smartindent: */