vxodbc/pgxalib.cpp

   1 /*------
   2  * Module:                      pgxalib.cpp
   3  *
   4  * Description:
   5  *              This module implements XA like routines
   6  *                      invoked from MSDTC process.
   7  *
   8  *              xa_open(), xa_close(), xa_commit(),
   9  *              xa_rollback() and xa_recover()
  10  *              are really invoked AFAIC.
  11  *-------
  12  */
  13
  14 #include <oleTx2xa.h>
  15 /*#define       _SLEEP_FOR_TEST_*/
  16 #include <sqlext.h>
  17 #include <odbcinst.h>
  18 #include <stdio.h>
  19 #include <string.h>
  20 #include <ctype.h>
  21 #include <process.h>
  22 #include <time.h>
  23
  24 #include <string>
  25 #include <map>
  26 #include <vector>
  27
  28
  29 EXTERN_C static void mylog(const char *fmt,...);
  30
  31 using namespace std;
  32
  33 class   XAConnection
  34 {
  35 private:
  36         string  connstr;
  37         HDBC    xaconn;
  38         vector<string>  qvec;
  39         int             pos;
  40 public:
  41         XAConnection(LPCTSTR str) : connstr(str), xaconn(NULL), pos(-1) {}
  42         ~XAConnection();
  43         HDBC    ActivateConnection(void);
  44         void    SetPos(int spos) {pos = spos;}
  45         HDBC    GetConnection(void) const {return xaconn;}
  46         vector<string>  &GetResultVec(void) {return qvec;}
  47         int     GetPos(void) {return pos;}
  48         const string &GetConnstr(void) {return connstr;}
  49 };
  50
  51 static class INIT_CRIT
  52 {
  53 private:
  54 public:
  55         bool                    cs_init;
  56         CRITICAL_SECTION        map_cs;
  57         CRITICAL_SECTION        mylog_cs;
  58         map<int, XAConnection>  xatab;
  59         FILE                    *LOGFP;
  60         HENV                    env;
  61
  62         INIT_CRIT() : LOGFP(NULL), env(NULL)
  63         {
  64                 InitializeCriticalSection(&map_cs);
  65                 InitializeCriticalSection(&mylog_cs);
  66                 cs_init = true;
  67         }
  68         ~INIT_CRIT()
  69         {
  70 // mylog("Leaving INIT_CRIT\n");
  71                 if (cs_init)
  72                 {
  73                         xatab.clear();
  74                         FreeEnv();
  75                         if (LOGFP)
  76                                 fclose(LOGFP);
  77                         DeleteCriticalSection(&mylog_cs);
  78                         DeleteCriticalSection(&map_cs);
  79                 }
  80         }
  81         void finalize()
  82         {
  83                 if (cs_init)
  84                 {
  85                         xatab.clear();
  86                         FreeEnv();
  87                         if (LOGFP)
  88                                 fclose(LOGFP);
  89                         LOGFP = NULL;
  90                         DeleteCriticalSection(&mylog_cs);
  91                         DeleteCriticalSection(&map_cs);
  92                         cs_init = false;
  93                 }
  94         }
  95         void FreeEnv()
  96         {
  97                 if (env)
  98                 {
  99                         SQLFreeHandle(SQL_HANDLE_ENV, env);
 100                         env = NULL;
 101                 }
 102         }
 103 } init_crit;
 104 #define MLOCK_ACQUIRE   EnterCriticalSection(&init_crit.map_cs)
 105 #define MLOCK_RELEASE   LeaveCriticalSection(&init_crit.map_cs)
 106
 107 static int dtclog = 0;
 108
 109 XAConnection::~XAConnection()
 110 {
 111         qvec.clear();
 112         if (xaconn)
 113                 SQLFreeHandle(SQL_HANDLE_DBC, xaconn);
 114 }
 115
 116 HDBC    XAConnection::ActivateConnection(void)
 117 {
 118         RETCODE ret;
 119
 120         MLOCK_ACQUIRE;
 121         if (!init_crit.env)
 122         {
 123                 ret = SQLAllocHandle(SQL_HANDLE_ENV, SQL_NULL_HANDLE, &init_crit.env);
 124                 if (SQL_SUCCESS != ret && SQL_SUCCESS_WITH_INFO != ret)
 125                         return NULL;
 126         }
 127         MLOCK_RELEASE;
 128         if (!xaconn)
 129         {
 130                 ret = SQLSetEnvAttr(init_crit.env, SQL_ATTR_ODBC_VERSION, (PTR) SQL_OV_ODBC3, 0);
 131                 ret = SQLAllocHandle(SQL_HANDLE_DBC, init_crit.env, &xaconn);
 132                 if (SQL_SUCCESS == ret || SQL_SUCCESS_WITH_INFO)
 133                 {
 134                         string  cstr = connstr;
 135                         if (dtclog)
 136                         {
 137                                 cstr += ";B2=";
 138                                 cstr += dtclog;
 139                         }
 140                         ret = SQLDriverConnect(xaconn, NULL,
 141                          (SQLCHAR *) cstr.c_str(), SQL_NTS, NULL, SQL_NULL_DATA, NULL, SQL_DRIVER_COMPLETE);
 142                         if (SQL_SUCCESS != ret && SQL_SUCCESS_WITH_INFO != ret)
 143                         {
 144                                 mylog("SQLDriverConnect return=%d\n", ret);
 145                                 SQLFreeHandle(SQL_HANDLE_DBC, xaconn);
 146                                 xaconn = NULL;
 147                         }
 148                 }
 149         }
 150         return xaconn;
 151 }
 152
 153 #define _BUILD_DLL_
 154 #ifdef _BUILD_DLL_
 155
 156 EXTERN_C {
 157
 158 #define DIRSEPARATOR    "\\"
 159 #define PG_BINARY_A     "ab"
 160 #define MYLOGDIR        "c:"
 161 #define MYLOGFILE       "mylog_"
 162
 163 static void
 164 generate_filename(const char *dirname, const char *prefix, char *filename)
 165 {
 166         int                     pid = 0;
 167
 168         pid = _getpid();
 169         if (dirname == 0 || filename == 0)
 170                 return;
 171
 172         strcpy(filename, dirname);
 173         strcat(filename, DIRSEPARATOR);
 174         if (prefix != 0)
 175                 strcat(filename, prefix);
 176         sprintf(filename, "%s%u%s", filename, pid, ".log");
 177         return;
 178 }
 179
 180 static void FreeEnv()
 181 {
 182         init_crit.FreeEnv();
 183 }
 184
 185 #define DTCLOGDIR       "c:\\pgdtclog"
 186 #include <direct.h>
 187
 188 static const char * const DBMSNAME = "PostgreSQL";
 189 static const char * const KEY_NAME = "MsdtcLog";
 190 static const char * const ODBCINST_INI = "ODBCINST.INI";
 191 INT_PTR FAR WINAPI GetMsdtclog()
 192 {
 193         char    temp[16];
 194
 195         SQLGetPrivateProfileString(DBMSNAME, KEY_NAME, "", temp, sizeof(temp), ODBCINST_INI);
 196         dtclog = atoi(temp);
 197         return dtclog;
 198 }
 199 INT_PTR FAR WINAPI SetMsdtclog(int dtclog)
 200 {
 201         char    temp[16];
 202
 203         sprintf(temp, "%d", dtclog);
 204         SQLWritePrivateProfileString(DBMSNAME, KEY_NAME, temp, ODBCINST_INI);
 205         return dtclog;
 206 }
 207
 208 static BOOL
 209 output_mylog()
 210 {
 211         return (0 != dtclog);
 212 }
 213
 214 static void
 215 mylog(const char *fmt,...)
 216 {
 217         va_list         args;
 218         char            filebuf[80];
 219         static  BOOL    init = TRUE;
 220         FILE    *logfp = init_crit.LOGFP;
 221
 222         if (!output_mylog())
 223                 return;
 224         EnterCriticalSection(&init_crit.mylog_cs);
 225         va_start(args, fmt);
 226
 227         if (init)
 228         {
 229                 if (!logfp)
 230                 {
 231                         generate_filename(MYLOGDIR, MYLOGFILE, filebuf);
 232                         logfp = fopen(filebuf, PG_BINARY_A);
 233                 }
 234                 if (!logfp)
 235                 {
 236                         generate_filename(DTCLOGDIR, MYLOGFILE, filebuf);
 237                         logfp = fopen(filebuf, PG_BINARY_A);
 238 #ifdef  WIN32
 239                         if (NULL == logfp)
 240                         {
 241                                 if (0 == _mkdir(DTCLOGDIR))
 242                                         logfp = fopen(filebuf, PG_BINARY_A);
 243                         }
 244 #endif /* WIN32 */
 245                 }
 246                 if (logfp)
 247                 {
 248                         setbuf(logfp, NULL);
 249                         init_crit.LOGFP = logfp;
 250                 }
 251         }
 252         init = FALSE;
 253         if (logfp)
 254         {
 255                 time_t  ntime;
 256                 char    ctim[128];
 257
 258                 time(&ntime);
 259                 strcpy(ctim, ctime(&ntime));
 260                 ctim[strlen(ctim) - 1] = '\0';
 261                 fprintf(logfp, "[%d.%d(%s)]", GetCurrentProcessId(), GetCurrentThreadId(), ctim);
 262                 vfprintf(logfp, fmt, args);
 263         }
 264         va_end(args);
 265         LeaveCriticalSection(&init_crit.mylog_cs);
 266 }
 267 static int      initialize_globals(void)
 268 {
 269         static  int     init = 1;
 270
 271         if (!init)
 272                 init = 0;
 273
 274         return 0;
 275 }
 276
 277 static void XatabClear(void);
 278 static void finalize_globals(void)
 279 {
 280         /* my(q)log is unavailable from here */
 281         mylog("DETACHING PROCESS\n");
 282         init_crit.finalize();
 283 }
 284
 285 HINSTANCE s_hModule;               /* Saved module handle. */
 286 /*      This is where the Driver Manager attaches to this Driver */
 287 BOOL    WINAPI
 288 DllMain(HANDLE hInst, ULONG ul_reason_for_call, LPVOID lpReserved)
 289 {
 290         WORD    wVersionRequested;
 291         WSADATA wsaData;
 292
 293         switch (ul_reason_for_call)
 294         {
 295                 case DLL_PROCESS_ATTACH:
 296                         s_hModule = (HINSTANCE) hInst;  /* Save for dialog boxes */
 297
 298                         /* Load the WinSock Library */
 299                         wVersionRequested = MAKEWORD(1, 1);
 300
 301                         if (WSAStartup(wVersionRequested, &wsaData))
 302                                 return FALSE;
 303
 304                         /* Verify that this is the minimum version of WinSock */
 305                         if (LOBYTE(wsaData.wVersion) != 1 ||
 306                                 HIBYTE(wsaData.wVersion) != 1)
 307                         {
 308                                 WSACleanup();
 309                                 return FALSE;
 310                         }
 311                         initialize_globals();
 312                         break;
 313
 314                 case DLL_THREAD_ATTACH:
 315                         break;
 316
 317                 case DLL_PROCESS_DETACH:
 318                         finalize_globals();
 319                         WSACleanup();
 320                         return TRUE;
 321
 322                 case DLL_THREAD_DETACH:
 323                         break;
 324
 325                 default:
 326                         break;
 327         }
 328         return TRUE;
 329 }
 330
 331 } /* end of EXTERN_C */
 332
 333 #endif /* _BUILD_DLL_ */
 334
 335 static void XatabClear(void)
 336 {
 337         init_crit.xatab.clear();
 338 }
 339
 340 static const char *XidToText(const XID &xid, char *rtext)
 341 {
 342         int     glen = xid.gtrid_length, blen = xid.bqual_length;
 343         int     i, j;
 344
 345         for (i = 0, j = 0; i < glen; i++, j += 2)
 346                 sprintf(rtext + j, "%02x", (unsigned char) xid.data[i]);
 347         strcat(rtext, "-"); j++;
 348         for (; i < glen + blen; i++, j += 2)
 349                 sprintf(rtext + j, "%02x", (unsigned char) xid.data[i]);
 350         return rtext;
 351 }
 352
 353 static int
 354 pg_hex2bin(const UCHAR *src, UCHAR *dst, int length)
 355 {
 356         UCHAR           chr;
 357         const UCHAR     *src_wk;
 358         UCHAR           *dst_wk;
 359         int             i, val;
 360         BOOL            HByte = TRUE;
 361
 362         for (i = 0, src_wk = src, dst_wk = dst; i < length; i++, src_wk++)
 363         {
 364                 chr = *src_wk;
 365                 if (!chr)
 366                         break;
 367                 if (chr >= 'a' && chr <= 'f')
 368                         val = chr - 'a' + 10;
 369                 else if (chr >= 'A' && chr <= 'F')
 370                         val = chr - 'A' + 10;
 371                 else
 372                         val = chr - '0';
 373                 if (HByte)
 374                         *dst_wk = (val << 4);
 375                 else
 376                 {
 377                         *dst_wk += val;
 378                         dst_wk++;
 379                 }
 380                 HByte = !HByte;
 381         }
 382         return length;
 383 }
 384
 385 static int      TextToXid(XID &xid, const char *rtext)
 386 {
 387         int     slen, glen, blen;
 388         char    *sptr;
 389
 390         slen = strlen(rtext);
 391         sptr = (char *)strchr(rtext, '-');
 392         if (sptr)
 393         {
 394                 glen = (int) (sptr - rtext);
 395                 blen = slen - glen - 1;
 396         }
 397         else
 398         {
 399                 glen = slen;
 400                 blen = 0;
 401         }
 402         xid.gtrid_length = glen / 2;
 403         xid.bqual_length = blen / 2;
 404         pg_hex2bin((const UCHAR *) rtext, (UCHAR *) &xid.data[0], glen);
 405         pg_hex2bin((const UCHAR *) sptr + 1, (UCHAR *) &xid.data[glen / 2], blen);
 406         return (glen + blen) / 2;
 407 }
 408
 409
 410 EXTERN_C static int __cdecl xa_open(char *xa_info, int rmid, long flags)
 411 {
 412         mylog("xa_open %s rmid=%d flags=%ld\n", xa_info, rmid, flags);
 413         GetMsdtclog();
 414         MLOCK_ACQUIRE;
 415         init_crit.xatab.insert(pair<int, XAConnection>(rmid, XAConnection(xa_info)));
 416         MLOCK_RELEASE;
 417         return  S_OK;
 418 }
 419 EXTERN_C static int __cdecl xa_close(char *xa_info, int rmid, long flags)
 420 {
 421         mylog("xa_close rmid=%d flags=%ld\n", rmid, flags);
 422         MLOCK_ACQUIRE;
 423         init_crit.xatab.erase(rmid);
 424         if (init_crit.xatab.size() == 0)
 425                 FreeEnv();
 426         GetMsdtclog();
 427         MLOCK_RELEASE;
 428         return XA_OK;
 429 }
 430
 431 //
 432 //      Dummy implementation (not called from MSDTC).
 433 //
 434 EXTERN_C static int __cdecl xa_start(XID *xid, int rmid, long flags)
 435 {
 436         char    pgxid[258];
 437
 438         XidToText(*xid, pgxid);
 439         mylog("xa_start %s rmid=%d flags=%ld\n", pgxid, rmid, flags);
 440         init_crit.xatab.find(rmid)->second.ActivateConnection();
 441         return XA_OK;
 442 }
 443 //
 444 //      Dummy implementation (not called from MSDTC).
 445 //
 446 EXTERN_C static int __cdecl xa_end(XID *xid, int rmid, long flags)
 447 {
 448         char    pgxid[258];
 449
 450         XidToText(*xid, pgxid);
 451         mylog("xa_end %s rmid=%d flags=%ld\n", pgxid, rmid, flags);
 452         return XA_OK;
 453 }
 454
 455 EXTERN_C static int __cdecl xa_rollback(XID *xid, int rmid, long flags)
 456 {
 457         int     rmcode = XAER_RMERR;
 458         char    pgxid[258];
 459
 460         XidToText(*xid, pgxid);
 461         mylog("xa_rollback %s rmid=%d flags=%ld\n", pgxid, rmid, flags);
 462         map<int, XAConnection>::iterator p;
 463         p = init_crit.xatab.find(rmid);
 464         if (p != init_crit.xatab.end())
 465         {
 466                 HDBC    conn = p->second.ActivateConnection();
 467                 if (conn)
 468                 {
 469                         SQLCHAR cmdmsg[512], sqlstate[8];
 470                         HSTMT   stmt;
 471                         RETCODE ret;
 472
 473                         ret = SQLAllocHandle(SQL_HANDLE_STMT, conn, &stmt);
 474                         if (SQL_SUCCESS != ret && SQL_SUCCESS_WITH_INFO != ret)
 475                         {
 476                                 mylog("Statement allocation error\n");
 477                                 return rmcode;
 478                         }
 479                         _snprintf((char *) cmdmsg, sizeof(cmdmsg), "ROLLBACK PREPARED '%s'", pgxid);
 480                         ret = SQLExecDirect(stmt, (SQLCHAR *) cmdmsg, SQL_NTS);
 481                         switch (ret)
 482                         {
 483                                 case SQL_SUCCESS:
 484                                 case SQL_SUCCESS_WITH_INFO:
 485                                         rmcode = XA_OK;
 486                                         break;
 487                                 case SQL_ERROR:
 488                                         SQLGetDiagRec(SQL_HANDLE_STMT, stmt,
 489                                                 1, sqlstate, NULL, cmdmsg,
 490                                                 sizeof(cmdmsg), NULL);
 491                                         mylog("xa_commit error %s '%s'\n", sqlstate, cmdmsg);
 492                                         if (_stricmp((char *) sqlstate, "42704") == 0)
 493                                                 rmcode = XA_HEURHAZ;
 494                                         break;
 495                         }
 496                         SQLFreeHandle(SQL_HANDLE_STMT, stmt);
 497                 }
 498         }
 499         return rmcode;
 500 }
 501 //
 502 //      Dummy implementation (not called from MSDTC).
 503 //      ANway it's almost impossible to implement this routine properly.
 504 //
 505 EXTERN_C static int __cdecl xa_prepare(XID *xid, int rmid, long flags)
 506 {
 507         char    pgxid[258];
 508
 509         XidToText(*xid, pgxid);
 510         mylog("xa_prepare %s rmid=%d\n", pgxid, rmid);
 511 #ifdef  _SLEEP_FOR_TEST_
 512         Sleep(2000);
 513 #endif  /* _SLEEP_FOR_TEST_ */
 514         map<int, XAConnection>::iterator p;
 515         p = init_crit.xatab.find(rmid);
 516         if (p != init_crit.xatab.end())
 517         {
 518                 HDBC    conn = p->second.GetConnection();
 519                 if (conn)
 520                 {
 521                 }
 522         }
 523         return XAER_RMERR;
 524 }
 525 EXTERN_C static int __cdecl xa_commit(XID *xid, int rmid, long flags)
 526 {
 527         int     rmcode = XAER_RMERR;
 528         char    pgxid[258];
 529
 530         XidToText(*xid, pgxid);
 531         mylog("xa_commit %s rmid=%d flags=%ld\n", pgxid, rmid, flags);
 532 #ifdef  _SLEEP_FOR_TEST_
 533         Sleep(2000);
 534 #endif  /* _SLEEP_FOR_TEST_ */
 535         map<int, XAConnection>::iterator p;
 536         p = init_crit.xatab.find(rmid);
 537         if (p != init_crit.xatab.end())
 538         {
 539                 HDBC    conn = p->second.ActivateConnection();
 540                 if (conn)
 541                 {
 542                         SQLCHAR cmdmsg[512], sqlstate[8];
 543                         HSTMT   stmt;
 544                         RETCODE ret;
 545
 546                         SQLAllocHandle(SQL_HANDLE_STMT, conn, &stmt);
 547                         _snprintf((char *) cmdmsg, sizeof(cmdmsg), "COMMIT PREPARED '%s'", pgxid);
 548                         ret = SQLExecDirect(stmt, (SQLCHAR *) cmdmsg, SQL_NTS);
 549                         switch (ret)
 550                         {
 551                                 case SQL_SUCCESS:
 552                                 case SQL_SUCCESS_WITH_INFO:
 553                                         rmcode = XA_OK;
 554                                         break;
 555                                 case SQL_ERROR:
 556                                         SQLGetDiagRec(SQL_HANDLE_STMT, stmt,
 557                                                 1, sqlstate, NULL, cmdmsg,
 558                                                 sizeof(cmdmsg), NULL);
 559                                         if (_stricmp((char *) sqlstate, "42704") == 0)
 560                                                 rmcode = XA_HEURHAZ;
 561                                         break;
 562                         }
 563                         SQLFreeHandle(SQL_HANDLE_STMT, stmt);
 564                 }
 565         }
 566         return rmcode;
 567 }
 568 EXTERN_C static int __cdecl xa_recover(XID *xids, long count, int rmid, long flags)
 569 {
 570         int     rmcode = XAER_RMERR, rcount;
 571
 572         mylog("xa_recover rmid=%d count=%d flags=%ld\n", rmid, count, flags);
 573         map<int, XAConnection>::iterator p;
 574         p = init_crit.xatab.find(rmid);
 575         if (p == init_crit.xatab.end())
 576                 return rmcode;
 577         HDBC    conn = p->second.ActivateConnection();
 578         if (!conn)
 579                 return rmcode;
 580         vector<string>  &vec = p->second.GetResultVec();
 581         int     pos = p->second.GetPos();
 582         if ((flags & TMSTARTRSCAN) != 0)
 583         {
 584                 HSTMT   stmt;
 585                 RETCODE ret;
 586                 char    buf[512];
 587
 588                 vec.clear();
 589                 SQLAllocHandle(SQL_HANDLE_STMT, conn, &stmt);
 590                 ret = SQLExecDirect(stmt, (SQLCHAR *) "select gid from pg_prepared_xacts", SQL_NTS);
 591                 if (SQL_SUCCESS != ret && SQL_SUCCESS_WITH_INFO != ret)
 592                 {
 593                         SQLFreeHandle(SQL_HANDLE_STMT, stmt);
 594                         pos = -1;
 595                         goto onExit;
 596                 }
 597                 SQLBindCol(stmt, 1, SQL_C_CHAR, buf, sizeof(buf), NULL);
 598                 ret = SQLFetch(stmt);
 599                 while (SQL_NO_DATA_FOUND != ret)
 600                 {
 601                         vec.push_back(buf);
 602                         ret = SQLFetch(stmt);
 603                 }
 604                 SQLFreeHandle(SQL_HANDLE_STMT, stmt);
 605                 pos = 0;
 606         }
 607         rcount = vec.size();
 608         rmcode = rcount - pos;
 609         if (rmcode > count)
 610                 rmcode = count;
 611         for (int i = 0; i < rmcode; i++, pos++)
 612                 TextToXid(xids[i], vec[pos].c_str());
 613
 614         if ((flags & TMENDRSCAN) != 0)
 615         {
 616                 vec.clear();
 617                 pos = -1;
 618         }
 619         mylog("return count=%d\n", rmcode);
 620 onExit:
 621         p->second.SetPos(pos);
 622         return rmcode;
 623 }
 624
 625 //
 626 //      I'm not sure if this is invoked from MSDTC
 627 //      Anyway there's nothing to do with it.
 628 //
 629 EXTERN_C static int __cdecl xa_forget(XID *xid, int rmid, long flags)
 630 {
 631         char    pgxid[258];
 632
 633         XidToText(*xid, pgxid);
 634         mylog("xa_forget %s rmid=%d\n", pgxid, rmid);
 635         return XA_OK;
 636 }
 637 //
 638 //      I'm not sure if this can be invoked from MSDTC.
 639 //
 640 EXTERN_C static int __cdecl xa_complete(int *handle, int *retval, int rmid, long flags)
 641 {
 642         mylog("xa_complete rmid=%d\n", rmid);
 643         return XA_OK;
 644 }
 645
 646 EXTERN_C static xa_switch_t     xapsw = { "psotgres_xa", TMNOMIGRATE,
 647                 0, xa_open, xa_close, xa_start, xa_end, xa_rollback,
 648                 xa_prepare, xa_commit, xa_recover, xa_forget,
 649                 xa_complete};
 650
 651 EXTERN_C HRESULT __cdecl   GetXaSwitch (XA_SWITCH_FLAGS  XaSwitchFlags,
 652                 xa_switch_t **  ppXaSwitch)
 653 {
 654         mylog("GetXaSwitch called\n");
 655
 656         GetMsdtclog();
 657         *ppXaSwitch = &xapsw;
 658         return  S_OK;
 659 }