« back to all changes in this revision
Viewing changes to msdtc_enlist.cpp
- browse files at revision 19
- compare with another revision
- download diff
- download tarball
- view history from revision 19
- Committer: Package Import Robot
- Author(s): Christoph Berg
- Date: 2014-05-29 23:17:25 UTC
- mfrom: (16.1.8 sid)
- Revision ID: package-import@ubuntu.com-20140529231725-nhpolx85545e4rk8
Tags: 1:09.03.0300-1
* New upstream release.
* Patch bogus expected output of test catalogfunctions.
* Set team as maintainer.
* Bump to dh 9.
* Use /usr/share/cdbs/1/rules/autoreconf.mk. Closes: #744650.
* Patch bogus expected output of test catalogfunctions.
* Set team as maintainer.
* Bump to dh 9.
* Use /usr/share/cdbs/1/rules/autoreconf.mk. Closes: #744650.
- files added:
- .pc/catalogfunctions-expected-output
- .pc/catalogfunctions-expected-output/test
- .pc/catalogfunctions-expected-output/test/expected
- installer/psqlodbc-setup
- winbuild
- files removed:
- .pc/test-crlf
- .pc/test-crlf/test
- .pc/test-crlf/test/expected
- .pc/test-crlf/test/sql
- .pc/testsuite.patch
- .pc/testsuite.patch/test
- .pc/testsuite.patch/test/src
- files modified:
- .pc/applied-patches
added
removed
msdtc_enlist.cpp
1
/*------
2
* Module: msdtc_enlist.cpp
3
*
4
* Description:
5
* This module contains routines related to
6
* the enlistment in MSDTC.
7
*
8
*-------
9
*/
10
11
#ifdef _HANDLE_ENLIST_IN_DTC_
12
13
#undef _MEMORY_DEBUG_
14
#ifndef _WIN32_WINNT
15
#define _WIN32_WINNT 0x0400
16
#endif /* _WIN32_WINNT */
17
18
#define WIN32_LEAN_AND_MEAN
19
#include <oleTx2xa.h>
20
#include <XOLEHLP.h>
21
/*#include <Txdtc.h>*/
22
#include "connection.h"
23
24
/*#define _SLEEP_FOR_TEST_*/
25
#include <stdio.h>
26
#include <string.h>
27
#include <ctype.h>
28
#include <process.h>
29
#include <map>
30
#ifndef WIN32
31
#include <errno.h>
32
#endif /* WIN32 */
33
34
#include "qresult.h"
35
#include "dlg_specific.h"
36
37
#include "pgapifunc.h"
38
#include "pgenlist.h"
39
40
EXTERN_C {
41
HINSTANCE s_hModule; /* Saved module handle. */
42
}
43
/* This is where the Driver Manager attaches to this Driver */
44
BOOL WINAPI
45
DllMain(HANDLE hInst, ULONG ul_reason_for_call, LPVOID lpReserved)
46
{
47
switch (ul_reason_for_call)
48
{
49
case DLL_PROCESS_ATTACH:
50
s_hModule = (HINSTANCE) hInst; /* Save for dialog boxes
51
*/
52
break;
53
}
54
return TRUE;
55
}
56
57
static class INIT_CRIT
58
{
59
public:
60
CRITICAL_SECTION life_cs;
61
CRITICAL_SECTION map_cs;
62
INIT_CRIT() {
63
InitializeCriticalSection(&life_cs);
64
InitializeCriticalSection(&map_cs);
65
}
66
~INIT_CRIT() {
67
DeleteCriticalSection(&life_cs);
68
DeleteCriticalSection(&map_cs);
69
}
70
} init_crit;
71
#define LIFELOCK_ACQUIRE EnterCriticalSection(&init_crit.life_cs)
72
#define LIFELOCK_RELEASE LeaveCriticalSection(&init_crit.life_cs)
73
#define MLOCK_ACQUIRE EnterCriticalSection(&init_crit.map_cs)
74
#define MLOCK_RELEASE LeaveCriticalSection(&init_crit.map_cs)
75
76
77
static const char *XidToText(const XID &xid, char *rtext)
78
{
79
int glen = xid.gtrid_length, blen = xid.bqual_length;
80
int i, j;
81
82
for (i = 0, j = 0; i < glen; i++, j += 2)
83
sprintf(rtext + j, "%02x", (unsigned char) xid.data[i]);
84
strcat(rtext, "-"); j++;
85
for (; i < glen + blen; i++, j += 2)
86
sprintf(rtext + j, "%02x", (unsigned char) xid.data[i]);
87
return rtext;
88
}
89
90
static LONG g_cComponents = 0;
91
static LONG g_cServerLocks = 0;
92
93
//
94
// �ȉ���ITransactionResourceAsync�I�u�W�F�N�g�͔C�ӂ̃X���b�h����
95
// ���R�ɃA�N�Z�X�\�Ȃ悤�Ɏ�������B�eRequest�̌��ʂ�Ԃ����߂�
96
// �g�p����ITransactionEnlistmentAsync�C���^�[�t�F�C�X�����̂悤��
97
// ��������Ă���i�Ǝv����A���L�Q�Ɓj�̂ŌĂяo����COM�̃A�p�[
98
// �g�����g���ӎ�����(CoMarshalInterThreadInterfaceInStream/CoGetIn
99
// terfaceAndReleaseStream���g�p����j�K�v�͂Ȃ��B
100
// ����DLL���Ŏg�p����ITransactionResourceAsync��ITransactionEnlist
101
// mentAsync�̃C���^�[�t�F�C�X�|�C���^�[�͔C�ӂ̃X���b�h���璼�ڎg�p
102
// ���邱�Ƃ��ł���B
103
//
104
105
// OLE Transactions Standard
106
//
107
// OLE Transactions is the Microsoft interface standard for transaction
108
// management. Applications use OLE Transactions-compliant interfaces to
109
// initiate, commit, abort, and inquire about transactions. Resource
110
// managers use OLE Transactions-compliant interfaces to enlist in
111
// transactions, to propagate transactions to other resource managers,
112
// to propagate transactions from process to process or from system to
113
// system, and to participate in the two-phase commit protocol.
114
//
115
// The Microsoft DTC system implements most OLE Transactions-compliant
116
// objects, interfaces, and methods. Resource managers that wish to use
117
// OLE Transactions must implement some OLE Transactions-compliant objects,
118
// interfaces, and methods.
119
//
120
// The OLE Transactions specification is based on COM but it differs in the
121
// following respects:
122
//
123
// OLE Transactions objects cannot be created using the COM CoCreate APIs.
124
// References to OLE Transactions objects are always direct. Therefore,
125
// no proxies or stubs are created for inter-apartment, inter-process,
126
// or inter-node calls and OLE Transactions references cannot be marshaled
127
// using standard COM marshaling.
128
// All references to OLE Transactions objects and their sinks are completely
129
// free threaded and cannot rely upon COM concurrency control models.
130
// For example, you cannot pass a reference to an IResourceManagerSink
131
// interface on a single-threaded apartment and expect the callback to occur
132
// only on the same single-threaded apartment.
133
134
/*#define _LOCK_DEBUG_ */
135
class IAsyncPG : public ITransactionResourceAsync
136
{
137
friend class AsyncThreads;
138
private:
139
IDtcToXaHelperSinglePipe *helper;
140
DWORD RMCookie;
141
ConnectionClass *conn;
142
ConnectionClass *xaconn;
143
LONG refcnt;
144
CRITICAL_SECTION as_spin; // to make this object Both
145
CRITICAL_SECTION as_exec; // to make this object Both
146
XID xid;
147
bool prepared;
148
HANDLE eThread[3];
149
HRESULT prepare_result;
150
bool requestAccepted;
151
HRESULT commit_result;
152
#ifdef _LOCK_DEBUG_
153
int spin_cnt;
154
int cs_cnt;
155
#endif /* _LOCK_DEBUG_ */
156
157
public:
158
enum {
159
PrepareExec = 0
160
,CommitExec
161
,AbortExec
162
};
163
164
ITransactionEnlistmentAsync *enlist;
165
166
HRESULT STDMETHODCALLTYPE QueryInterface(REFIID iid, void ** ppvObject);
167
ULONG STDMETHODCALLTYPE AddRef(void);
168
ULONG STDMETHODCALLTYPE Release(void);
169
170
HRESULT STDMETHODCALLTYPE PrepareRequest(BOOL fRetaining,
171
DWORD grfRM,
172
BOOL fWantMoniker,
173
BOOL fSinglePhase);
174
HRESULT STDMETHODCALLTYPE CommitRequest(DWORD grfRM, XACTUOW * pNewUOW);
175
HRESULT STDMETHODCALLTYPE AbortRequest(BOID * pboidReason,
176
BOOL fRetaining,
177
XACTUOW * pNewUOW);
178
HRESULT STDMETHODCALLTYPE TMDown(void);
179
180
IAsyncPG();
181
void SetHelper(IDtcToXaHelperSinglePipe *pHelper, DWORD dwRMCookie) {helper = pHelper; RMCookie = dwRMCookie;}
182
183
HRESULT RequestExec(DWORD type, HRESULT res);
184
HRESULT ReleaseConnection(void);
185
void SetConnection(ConnectionClass *sconn) {SLOCK_ACQUIRE(); conn = sconn; SLOCK_RELEASE();}
186
void SetXid(const XID *ixid) {SLOCK_ACQUIRE(); xid = *ixid; SLOCK_RELEASE();}
187
private:
188
~IAsyncPG();
189
#ifdef _LOCK_DEBUG_
190
void SLOCK_ACQUIRE() {forcelog("SLOCK_ACQUIRE %d\n", spin_cnt); EnterCriticalSection(&as_spin); spin_cnt++;}
191
void SLOCK_RELEASE() {forcelog("SLOCK_RELEASE=%d\n", spin_cnt); LeaveCriticalSection(&as_spin); spin_cnt--;}
192
#else
193
void SLOCK_ACQUIRE() {EnterCriticalSection(&as_spin);}
194
void SLOCK_RELEASE() {LeaveCriticalSection(&as_spin);}
195
#endif /* _LOCK_DEBUG_ */
196
void ELOCK_ACQUIRE() {EnterCriticalSection(&as_exec);}
197
void ELOCK_RELEASE() {LeaveCriticalSection(&as_exec);}
198
ConnectionClass *getLockedXAConn(void);
199
ConnectionClass *generateXAConn(bool spinAcquired);
200
void SetPrepareResult(HRESULT res) {SLOCK_ACQUIRE(); prepared = true; prepare_result = res; SLOCK_RELEASE();}
201
void SetDone(HRESULT);
202
void Reset_eThread(int idx) {SLOCK_ACQUIRE(); eThread[idx] = NULL; SLOCK_RELEASE();}
203
void Wait_pThread(bool slock_hold);
204
void Wait_cThread(bool slock_hold, bool once);
205
};
206
207
//
208
// For thread control.
209
//
210
class AsyncWait {
211
private:
212
IAsyncPG *obj;
213
DWORD type;
214
int waiting_count;
215
public:
216
AsyncWait(IAsyncPG *async, DWORD itype) : obj(async), type(itype), waiting_count(0) {}
217
AsyncWait(const AsyncWait &a_th) : obj(a_th.obj), type(a_th.type), waiting_count(a_th.waiting_count) {}
218
~AsyncWait() {}
219
IAsyncPG *GetObj() const {return obj;}
220
DWORD GetType() const {return type;}
221
int WaitCount() const {return waiting_count;}
222
int StartWaiting() {return ++waiting_count;}
223
int StopWaiting() {return --waiting_count;}
224
};
225
//
226
// List of threads invoked from IAsyncPG objects.
227
//
228
class AsyncThreads {
229
private:
230
static std::map <HANDLE, AsyncWait> th_list;
231
public:
232
static void insert(HANDLE, IAsyncPG *, DWORD);
233
static void CleanupThreads(DWORD millisecond);
234
static bool WaitThread(IAsyncPG *, DWORD type, DWORD millisecond);
235
};
236
237
238
#define SYNC_AUTOCOMMIT(conn) (SQL_AUTOCOMMIT_OFF != conn->connInfo.autocommit_public ? (conn->transact_status |= CONN_IN_AUTOCOMMIT) : (conn->transact_status &= ~CONN_IN_AUTOCOMMIT))
239
240
IAsyncPG::IAsyncPG(void) : helper(NULL), RMCookie(0), enlist(NULL), conn(NULL), xaconn(NULL), refcnt(1), prepared(false), requestAccepted(false)
241
{
242
InterlockedIncrement(&g_cComponents);
243
InitializeCriticalSection(&as_spin);
244
InitializeCriticalSection(&as_exec);
245
eThread[0] = eThread[1] = eThread[2] = NULL;
246
memset(&xid, 0, sizeof(xid));
247
#ifdef _LOCK_DEBUG_
248
spin_cnt = 0;
249
cs_cnt = 0;
250
#endif /* _LOCK_DEBUG_ */
251
}
252
253
//
254
// invoked from *delete*.
255
// When entered ELOCK -> LIFELOCK -> SLOCK are acquired
256
// and they are released.
257
//
258
IAsyncPG::~IAsyncPG(void)
259
{
260
ConnectionClass *fconn = NULL;
261
262
if (conn)
263
{
264
conn->asdum = NULL;
265
conn = NULL;
266
}
267
if (xaconn)
268
{
269
fconn = xaconn;
270
xaconn->asdum = NULL;
271
xaconn = NULL;
272
}
273
SLOCK_RELEASE();
274
LIFELOCK_RELEASE;
275
if (fconn)
276
PGAPI_FreeConnect((HDBC) fconn);
277
DeleteCriticalSection(&as_spin);
278
ELOCK_RELEASE();
279
DeleteCriticalSection(&as_exec);
280
InterlockedDecrement(&g_cComponents);
281
}
282
HRESULT STDMETHODCALLTYPE IAsyncPG::QueryInterface(REFIID riid, void ** ppvObject)
283
{
284
forcelog("%x QueryInterface called\n", this);
285
if (riid == IID_IUnknown || riid == IID_ITransactionResourceAsync)
286
{
287
*ppvObject = this;
288
AddRef();
289
return S_OK;
290
}
291
*ppvObject = NULL;
292
return E_NOINTERFACE;
293
}
294
//
295
// acquire/releases SLOCK.
296
//
297
ULONG STDMETHODCALLTYPE IAsyncPG::AddRef(void)
298
{
299
mylog("%x->AddRef called\n", this);
300
SLOCK_ACQUIRE();
301
refcnt++;
302
SLOCK_RELEASE();
303
return refcnt;
304
}
305
//
306
// acquire/releases [ELOCK -> LIFELOCK -> ] SLOCK.
307
//
308
ULONG STDMETHODCALLTYPE IAsyncPG::Release(void)
309
{
310
mylog("%x->Release called refcnt=%d\n", this, refcnt);
311
SLOCK_ACQUIRE();
312
refcnt--;
313
if (refcnt <= 0)
314
{
315
SLOCK_RELEASE();
316
ELOCK_ACQUIRE();
317
LIFELOCK_ACQUIRE;
318
SLOCK_ACQUIRE();
319
if (refcnt <=0)
320
{
321
mylog("delete %x\n", this);
322
delete this;
323
return 0;
324
}
325
else
326
{
327
SLOCK_RELEASE();
328
LIFELOCK_RELEASE;
329
ELOCK_RELEASE();
330
}
331
}
332
else
333
SLOCK_RELEASE();
334
return refcnt;
335
}
336
337
//
338
// Acquire/release [MLOCK -> ] SLOCK.
339
//
340
void IAsyncPG::Wait_pThread(bool slock_hold)
341
{
342
mylog("Wait_pThread %d in\n", slock_hold);
343
HANDLE wThread;
344
int wait_idx = PrepareExec;
345
bool th_found;
346
if (!slock_hold)
347
SLOCK_ACQUIRE();
348
while (NULL != eThread[wait_idx])
349
{
350
wThread = eThread[wait_idx];
351
SLOCK_RELEASE();
352
th_found = AsyncThreads::WaitThread(this, wait_idx, 2000);
353
SLOCK_ACQUIRE();
354
if (th_found)
355
break;
356
}
357
if (!slock_hold)
358
SLOCK_RELEASE();
359
mylog("Wait_pThread out\n");
360
}
361
362
//
363
// Acquire/releases [MLOCK -> ] SLOCK.
364
//
365
void IAsyncPG::Wait_cThread(bool slock_hold, bool once)
366
{
367
HANDLE wThread;
368
int wait_idx;
369
bool th_found;
370
371
mylog("Wait_cThread %d,%d in\n", slock_hold, once);
372
if (!slock_hold)
373
SLOCK_ACQUIRE();
374
if (NULL != eThread[CommitExec])
375
wait_idx = CommitExec;
376
else
377
wait_idx = AbortExec;
378
while (NULL != eThread[wait_idx])
379
{
380
wThread = eThread[wait_idx];
381
SLOCK_RELEASE();
382
th_found = AsyncThreads::WaitThread(this, wait_idx, 2000);
383
SLOCK_ACQUIRE();
384
if (once || th_found)
385
break;
386
}
387
if (!slock_hold)
388
SLOCK_RELEASE();
389
mylog("Wait_cThread out\n");
390
}
391
392
/* Processing Prepare/Commit Request */
393
typedef
394
struct RequestPara {
395
DWORD type;
396
LPVOID lpr;
397
HRESULT res;
398
} RequestPara;
399
400
//
401
// Acquire/releases LIFELOCK -> SLOCK.
402
// may acquire/release ELOCK.
403
//
404
void IAsyncPG::SetDone(HRESULT res)
405
{
406
LIFELOCK_ACQUIRE;
407
SLOCK_ACQUIRE();
408
prepared = false;
409
requestAccepted = true;
410
commit_result = res;
411
if (conn || xaconn)
412
{
413
if (conn)
414
{
415
conn->asdum = NULL;
416
SYNC_AUTOCOMMIT(conn);
417
conn = NULL;
418
}
419
SLOCK_RELEASE();
420
LIFELOCK_RELEASE;
421
ELOCK_ACQUIRE();
422
if (xaconn)
423
{
424
xaconn->asdum = NULL;
425
PGAPI_FreeConnect(xaconn);
426
xaconn = NULL;
427
}
428
ELOCK_RELEASE();
429
}
430
else
431
{
432
SLOCK_RELEASE();
433
LIFELOCK_RELEASE;
434
}
435
}
436
437
//
438
// Acquire/releases [ELOCK -> LIFELOCK -> ] SLOCK.
439
//
440
ConnectionClass *IAsyncPG::generateXAConn(bool spinAcquired)
441
{
442
if (!spinAcquired)
443
SLOCK_ACQUIRE();
444
if (prepared && !xaconn)
445
{
446
SLOCK_RELEASE();
447
ELOCK_ACQUIRE();
448
LIFELOCK_ACQUIRE;
449
SLOCK_ACQUIRE();
450
if (prepared && !xaconn)
451
{
452
PGAPI_AllocConnect(conn->henv, (HDBC *) &xaconn);
453
memcpy(&xaconn->connInfo, &conn->connInfo, sizeof(ConnInfo));
454
conn->asdum = NULL;
455
SYNC_AUTOCOMMIT(conn);
456
conn = NULL;
457
SLOCK_RELEASE();
458
LIFELOCK_RELEASE;
459
CC_connect(xaconn, AUTH_REQ_OK, NULL);
460
}
461
else
462
{
463
SLOCK_RELEASE();
464
LIFELOCK_RELEASE;
465
}
466
ELOCK_RELEASE();
467
}
468
else
469
SLOCK_RELEASE();
470
return xaconn;
471
}
472
473
//
474
// [when entered]
475
// ELOCK is acquired.
476
//
477
// Acquire/releases SLOCK.
478
// Try to acquire CONNLOCK also.
479
//
480
// [on exit]
481
// ELOCK is kept acquired.
482
// If the return connection != NULL
483
// the CONNLOCK for the connection is acquired.
484
//
485
ConnectionClass *IAsyncPG::getLockedXAConn()
486
{
487
SLOCK_ACQUIRE();
488
if (!xaconn && conn && !CC_is_in_trans(conn))
489
{
490
if (TRY_ENTER_CONN_CS(conn))
491
{
492
if (CC_is_in_trans(conn))
493
{
494
LEAVE_CONN_CS(conn);
495
}
496
else
497
{
498
SLOCK_RELEASE();
499
return conn;
500
}
501
}
502
}
503
generateXAConn(true);
504
if (xaconn)
505
ENTER_CONN_CS(xaconn);
506
return xaconn;
507
}
508
509
//
510
// Acquire/release ELOCK [ -> MLOCK] -> SLOCK.
511
//
512
HRESULT IAsyncPG::RequestExec(DWORD type, HRESULT res)
513
{
514
HRESULT ret;
515
bool bReleaseEnlist = false;
516
ConnectionClass *econn;
517
QResultClass *qres;
518
char pgxid[258], cmd[512];
519
520
mylog("%x->RequestExec type=%d\n", this, type);
521
XidToText(xid, pgxid);
522
#ifdef _SLEEP_FOR_TEST_
523
/*Sleep(2000);*/
524
#endif /* _SLEEP_FOR_TEST_ */
525
ELOCK_ACQUIRE();
526
switch (type)
527
{
528
case PrepareExec:
529
if (XACT_S_SINGLEPHASE == res)
530
{
531
if (!CC_commit(conn))
532
res = E_FAIL;
533
bReleaseEnlist = true;
534
}
535
else if (E_FAIL != res)
536
{
537
snprintf(cmd, sizeof(cmd), "PREPARE TRANSACTION '%s'", pgxid);
538
qres = CC_send_query(conn, cmd, NULL, 0, NULL);
539
if (!QR_command_maybe_successful(qres))
540
res = E_FAIL;
541
QR_Destructor(qres);
542
}
543
ret = enlist->PrepareRequestDone(res, NULL, NULL);
544
SetPrepareResult(res);
545
break;
546
case CommitExec:
547
Wait_pThread(false);
548
if (E_FAIL != res)
549
{
550
econn = getLockedXAConn();
551
if (econn)
552
{
553
snprintf(cmd, sizeof(cmd), "COMMIT PREPARED '%s'", pgxid);
554
qres = CC_send_query(econn, cmd, NULL, 0, NULL);
555
if (!QR_command_maybe_successful(qres))
556
res = E_FAIL;
557
QR_Destructor(qres);
558
LEAVE_CONN_CS(econn);
559
}
560
}
561
SetDone(res);
562
ret = enlist->CommitRequestDone(res);
563
bReleaseEnlist = true;
564
break;
565
case AbortExec:
566
Wait_pThread(false);
567
if (prepared)
568
{
569
econn = getLockedXAConn();
570
if (econn)
571
{
572
snprintf(cmd, sizeof(cmd), "ROLLBACK PREPARED '%s'", pgxid);
573
qres = CC_send_query(econn, cmd, NULL, 0, NULL);
574
if (!QR_command_maybe_successful(qres))
575
res = E_FAIL;
576
QR_Destructor(qres);
577
LEAVE_CONN_CS(econn);
578
}
579
}
580
SetDone(res);
581
ret = enlist->AbortRequestDone(res);
582
bReleaseEnlist = true;
583
break;
584
default:
585
ret = -1;
586
}
587
if (bReleaseEnlist)
588
{
589
helper->ReleaseRMCookie(RMCookie, TRUE);
590
enlist->Release();
591
}
592
ELOCK_RELEASE();
593
mylog("%x->Done ret=%d\n", this, ret);
594
return ret;
595
}
596
597
//
598
// Acquire/releses [MLOCK -> ] SLOCK
599
// or [ELOCK -> LIFELOCK -> ] SLOCK.
600
//
601
HRESULT IAsyncPG::ReleaseConnection(void)
602
{
603
mylog("%x->ReleaseConnection\n", this);
604
ConnectionClass *iconn;
605
bool done = false;
606
607
SLOCK_ACQUIRE();
608
if (iconn = conn)
609
{
610
Wait_pThread(true);
611
if (NULL != eThread[CommitExec] || NULL != eThread[AbortExec] || requestAccepted)
612
{
613
if (prepared)
614
{
615
Wait_cThread(true, true);
616
if (!prepared)
617
done = true;
618
}
619
else
620
done = true;
621
if (done)
622
Wait_cThread(true, false);
623
}
624
if (conn && CONN_CONNECTED == conn->status && !done)
625
{
626
generateXAConn(true);
627
}
628
else
629
SLOCK_RELEASE();
630
}
631
else
632
SLOCK_RELEASE();
633
mylog("%x->ReleaseConnection exit\n", this);
634
return SQL_SUCCESS;
635
}
636
637
//
638
// Acquire/release [ELOCK -> ] [MLOCK -> ] SLOCK.
639
//
640
EXTERN_C static unsigned WINAPI DtcRequestExec(LPVOID para);
641
HRESULT STDMETHODCALLTYPE IAsyncPG::PrepareRequest(BOOL fRetaining, DWORD grfRM,
642
BOOL fWantMoniker, BOOL fSinglePhase)
643
{
644
HRESULT ret, res;
645
RequestPara *reqp;
646
647
mylog("%x PrepareRequest called grhRM=%d enl=%x\n", this, grfRM, enlist);
648
SLOCK_ACQUIRE();
649
if (0 != CC_get_errornumber(conn))
650
res = ret = E_FAIL;
651
else
652
{
653
ret = S_OK;
654
if (fSinglePhase)
655
{
656
res = XACT_S_SINGLEPHASE;
657
mylog("XACT is singlePhase\n");
658
}
659
else
660
res = S_OK;
661
}
662
SLOCK_RELEASE();
663
ELOCK_ACQUIRE();
664
#ifdef _SLEEP_FOR_TEST_
665
Sleep(2000);
666
#endif /* _SLEEP_FOR_TEST_ */
667
reqp = new RequestPara;
668
reqp->type = PrepareExec;
669
reqp->lpr = (LPVOID) this;
670
reqp->res = res;
671
AddRef();
672
HANDLE hThread = (HANDLE) _beginthreadex(NULL, 0, DtcRequestExec, reqp, 0, NULL);
673
if (NULL == hThread)
674
{
675
delete(reqp);
676
ret = E_FAIL;
677
}
678
else
679
{
680
AsyncThreads::insert(hThread, this, reqp->type);
681
}
682
ELOCK_RELEASE();
683
Release();
684
return ret;
685
}
686
//
687
// Acquire/release [ELOCK -> ] [MLOCK -> ] SLOCK.
688
//
689
HRESULT STDMETHODCALLTYPE IAsyncPG::CommitRequest(DWORD grfRM, XACTUOW * pNewUOW)
690
{
691
HRESULT res = S_OK, ret = S_OK;
692
RequestPara *reqp;
693
mylog("%x CommitRequest called grfRM=%d enl=%x\n", this, grfRM, enlist);
694
695
SLOCK_ACQUIRE();
696
if (!prepared)
697
ret = E_UNEXPECTED;
698
else if (S_OK != prepare_result)
699
ret = E_UNEXPECTED;
700
SLOCK_RELEASE();
701
if (S_OK != ret)
702
return ret;
703
AddRef();
704
ELOCK_ACQUIRE();
705
#ifdef _SLEEP_FOR_TEST_
706
Sleep(1000);
707
#endif /* _SLEEP_FOR_TEST_ */
708
reqp = new RequestPara;
709
reqp->type = CommitExec;
710
reqp->lpr = (LPVOID) this;
711
reqp->res = res;
712
enlist->AddRef();
713
HANDLE hThread = (HANDLE) _beginthreadex(NULL, 0, DtcRequestExec, reqp, 0, NULL);
714
if (NULL == hThread)
715
{
716
delete(reqp);
717
enlist->Release();
718
ret = E_FAIL;
719
}
720
else
721
{
722
AsyncThreads::insert(hThread, this, reqp->type);
723
}
724
mylog("CommitRequest ret=%d\n", ret);
725
requestAccepted = true;
726
ELOCK_RELEASE();
727
Release();
728
return ret;
729
}
730
//
731
// Acquire/release [ELOCK -> ] [MLOCK -> ] SLOCK.
732
//
733
HRESULT STDMETHODCALLTYPE IAsyncPG::AbortRequest(BOID * pboidReason, BOOL fRetaining,
734
XACTUOW * pNewUOW)
735
{
736
HRESULT res = S_OK, ret = S_OK;
737
RequestPara *reqp;
738
739
mylog("%x AbortRequest called\n", this);
740
AddRef();
741
ELOCK_ACQUIRE();
742
if (!prepared && conn)
743
CC_abort(conn);
744
reqp = new RequestPara;
745
reqp->type = AbortExec;
746
reqp->lpr = (LPVOID) this;
747
reqp->res = res;
748
enlist->AddRef();
749
HANDLE hThread = (HANDLE) _beginthreadex(NULL, 0, DtcRequestExec, reqp, 0, NULL);
750
if (NULL == hThread)
751
{
752
delete(reqp);
753
enlist->Release();
754
ret = E_FAIL;
755
}
756
else
757
{
758
AsyncThreads::insert(hThread, this, reqp->type);
759
}
760
mylog("AbortRequest ret=%d\n", ret);
761
requestAccepted = true;
762
ELOCK_RELEASE();
763
Release();
764
return ret;
765
}
766
HRESULT STDMETHODCALLTYPE IAsyncPG::TMDown(void)
767
{
768
forcelog("%x TMDown called\n", this);
769
return S_OK;
770
}
771
772
//
773
// Acquire/releases MLOCK -> SLOCK.
774
//
775
std::map<HANDLE, AsyncWait> AsyncThreads::th_list;
776
void AsyncThreads::insert(HANDLE th, IAsyncPG *obj, DWORD type)
777
{
778
if (!obj) return;
779
MLOCK_ACQUIRE;
780
th_list.insert(std::pair<HANDLE, AsyncWait>(th, AsyncWait(obj, type)));
781
obj->SLOCK_ACQUIRE();
782
obj->eThread[type] = th;
783
obj->SLOCK_RELEASE();
784
MLOCK_RELEASE;
785
}
786
787
//
788
// Acquire/releases MLOCK -> SLOCK.
789
//
790
bool AsyncThreads::WaitThread(IAsyncPG *obj, DWORD type, DWORD millisecond)
791
{
792
HANDLE th = NULL;
793
DWORD gtype;
794
bool typematch;
795
int wait_count;
796
797
MLOCK_ACQUIRE;
798
std::map<HANDLE, AsyncWait>::iterator p;
799
for (p = th_list.begin(); p != th_list.end(); p++)
800
{
801
gtype = p->second.GetType();
802
typematch = (gtype == type);
803
if (p->second.GetObj() == obj && typematch)
804
{
805
th = p->first;
806
break;
807
}
808
}
809
if (NULL == th)
810
{
811
MLOCK_RELEASE;
812
forcelog("WaitThread thread(%x, %d) not found\n", obj, type);
813
return false;
814
}
815
p->second.StartWaiting();
816
MLOCK_RELEASE;
817
818
DWORD ret = WaitForSingleObject(th, millisecond);
819
MLOCK_ACQUIRE;
820
wait_count = p->second.StopWaiting();
821
if (WAIT_OBJECT_0 == ret)
822
{
823
IAsyncPG *async = p->second.GetObj();
824
825
if (type >= 0 && type <= IAsyncPG::AbortExec)
826
async->Reset_eThread(type);
827
if (wait_count <= 0)
828
{
829
th_list.erase(th);
830
MLOCK_RELEASE;
831
CloseHandle(th);
832
if (type >= IAsyncPG::CommitExec)
833
{
834
async->Release();
835
}
836
}
837
else
838
MLOCK_RELEASE;
839
}
840
else
841
MLOCK_RELEASE;
842
return true;
843
}
844
845
void AsyncThreads::CleanupThreads(DWORD millisecond)
846
{
847
size_t msize;
848
DWORD nCount;
849
850
MLOCK_ACQUIRE;
851
if (msize = th_list.size(), msize <= 0)
852
{
853
MLOCK_RELEASE;
854
return;
855
}
856
857
mylog("CleanupThreads size=%d\n", msize);
858
HANDLE *hds = new HANDLE[msize];
859
std::map<HANDLE, AsyncWait>::iterator p;
860
for (p = th_list.begin(), nCount = 0; p != th_list.end(); p++)
861
{
862
hds[nCount++] = p->first;
863
p->second.StartWaiting();
864
}
865
MLOCK_RELEASE;
866
int i;
867
while (nCount > 0)
868
{
869
DWORD ret = WaitForMultipleObjects(nCount, hds, 0, millisecond);
870
if (ret >= nCount)
871
break;
872
HANDLE th = hds[ret];
873
MLOCK_ACQUIRE;
874
p = th_list.find(th);
875
if (p != th_list.end())
876
{
877
int wait_count = p->second.StopWaiting();
878
DWORD type = p->second.GetType();
879
IAsyncPG * async = p->second.GetObj();
880
881
if (type >= IAsyncPG::PrepareExec && type <= IAsyncPG::AbortExec)
882
async->Reset_eThread(type);
883
if (wait_count <= 0)
884
{
885
th_list.erase(th);
886
MLOCK_RELEASE;
887
CloseHandle(th);
888
if (type >= IAsyncPG::CommitExec)
889
{
890
async->Release();
891
}
892
}
893
else
894
MLOCK_RELEASE;
895
}
896
else
897
MLOCK_RELEASE;
898
for (i = ret; i < (int) nCount - 1; i++)
899
hds[i] = hds[i + 1];
900
nCount--;
901
}
902
for (i = 0; i < (int) nCount; i++)
903
{
904
p = th_list.find(hds[i]);
905
if (p != th_list.end())
906
p->second.StopWaiting();
907
}
908
delete [] hds;
909
}
910
911
912
EXTERN_C static unsigned WINAPI DtcRequestExec(LPVOID para)
913
{
914
RequestPara *reqp = (RequestPara *) para;
915
DWORD type = reqp->type;
916
IAsyncPG *async = (IAsyncPG *) reqp->lpr;
917
HRESULT res = reqp->res, ret;
918
919
mylog("DtcRequestExec type=%d", reqp->type);
920
delete(reqp);
921
ret = async->RequestExec(type, res);
922
mylog(" Done ret=%d\n", ret);
923
return ret;
924
}
925
926
CSTR regKey = "SOFTWARE\\Microsoft\\MSDTC\\XADLL";
927
928
RETCODE static EnlistInDtc_1pipe(ConnectionClass *conn, ITransaction *pTra, ITransactionDispenser *pDtc)
929
{
930
CSTR func = "EnlistInDtc_1pipe";
931
static IDtcToXaHelperSinglePipe *pHelper = NULL;
932
ITransactionResourceAsync *pRes = NULL;
933
IAsyncPG *asdum;
934
HRESULT res;
935
bool retry, errset;
936
DWORD dwRMCookie;
937
XID xid;
938
939
if (!pHelper)
940
{
941
res = pDtc->QueryInterface(IID_IDtcToXaHelperSinglePipe, (void **) &pHelper);
942
if (res != S_OK || !pHelper)
943
{
944
forcelog("DtcToXaHelperSingelPipe get error %d\n", res);
945
pHelper = NULL;
946
return SQL_ERROR;
947
}
948
}
949
res = (NULL != (asdum = new IAsyncPG)) ? S_OK : E_FAIL;
950
if (S_OK != res)
951
{
952
mylog("CoCreateInstance error %d\n", res);
953
return SQL_ERROR;
954
}
955
956
mylog("dllname=%s dsn=%s\n", GetXaLibName(), conn->connInfo.dsn); res = 0;
957
retry = false;
958
errset = false;
959
ConnInfo *ci = &(conn->connInfo);
960
char dtcname[1024];
961
snprintf(dtcname, sizeof(dtcname), "DRIVER={%s};SERVER=%s;PORT=%s;DATABASE=%s;UID=%s;PWD=%s;" ABBR_SSLMODE "=%s",
962
ci->drivername, ci->server, ci->port, ci->database, ci->username, ci->password, ci->sslmode);
963
do {
964
res = pHelper->XARMCreate(dtcname, (char *) GetXaLibName(), &dwRMCookie);
965
if (S_OK == res)
966
break;
967
mylog("XARMCreate error code=%x\n", res);
968
if (XACT_E_XA_TX_DISABLED == res)
969
{
970
CC_set_error(conn, CONN_UNSUPPORTED_OPTION, "XARMcreate error:Please enable XA transaction in MSDTC security configuration", func);
971
errset = true;
972
}
973
else if (!retry)
974
{
975
LONG ret;
976
HKEY sKey;
977
DWORD rSize;
978
979
ret = ::RegOpenKeyEx(HKEY_LOCAL_MACHINE, regKey, 0, KEY_QUERY_VALUE | KEY_SET_VALUE, &sKey);
980
if (ERROR_SUCCESS != ret)
981
ret = ::RegCreateKeyEx(HKEY_LOCAL_MACHINE, regKey, 0, NULL, REG_OPTION_NON_VOLATILE, KEY_ALL_ACCESS, NULL, &sKey, NULL);
982
if (ERROR_SUCCESS == ret)
983
{
984
switch (ret = ::RegQueryValueEx(sKey, "XADLL", NULL, NULL, NULL, &rSize))
985
{
986
case ERROR_SUCCESS:
987
if (rSize > 0)
988
break;
989
default:
990
ret = ::RegSetValueEx(sKey, GetXaLibName(), 0, REG_SZ,
991
(CONST BYTE *) GetXaLibPath(), (DWORD) strlen(GetXaLibPath()) + 1);
992
if (ERROR_SUCCESS == ret)
993
{
994
retry = true;
995
continue; // retry
996
}
997
CC_set_error(conn, CONN_UNSUPPORTED_OPTION, "XARMCreate error:Please register HKLM\\SOFTWARE\\Microsoft\\MSDTC\\XADLL", func);
998
break;
999
}
1000
::RegCloseKey(sKey);
1001
}
1002
}
1003
if (!errset)
1004
CC_set_error(conn, CONN_UNSUPPORTED_OPTION, "MSDTC XARMCreate error", func);
1005
return SQL_ERROR;
1006
} while (1);
1007
res = pHelper->ConvertTridToXID((DWORD *) pTra, dwRMCookie, &xid);
1008
if (res != S_OK)
1009
{
1010
mylog("ConvertTridToXid error %d\n", res);
1011
return SQL_ERROR;
1012
}
1013
{
1014
char pgxid[258];
1015
XidToText(xid, pgxid);
1016
mylog("ConvertTridToXID -> %s\n", pgxid);
1017
}
1018
asdum->SetXid(&xid);
1019
/* Create an IAsyncPG instance by myself */
1020
/* DLLGetClassObject(GUID_IAsyncPG, IID_ITransactionResourceAsync, (void **) &asdum); */
1021
1022
asdum->SetHelper(pHelper, dwRMCookie);
1023
res = pHelper->EnlistWithRM(dwRMCookie, pTra, asdum, &asdum->enlist);
1024
if (res != S_OK)
1025
{
1026
mylog("EnlistWithRM error %d\n", res);
1027
pHelper->ReleaseRMCookie(dwRMCookie, TRUE);
1028
return SQL_ERROR;
1029
}
1030
1031
mylog("asdum=%p start transaction\n", asdum);
1032
CC_set_autocommit(conn, FALSE);
1033
asdum->SetConnection(conn);
1034
conn->asdum = asdum;
1035
1036
return SQL_SUCCESS;
1037
}
1038
1039
1040
EXTERN_C RETCODE EnlistInDtc(ConnectionClass *conn, void *pTra, int method)
1041
{
1042
static ITransactionDispenser *pDtc = NULL;
1043
1044
if (!pTra)
1045
{
1046
IAsyncPG *asdum = (IAsyncPG *) conn->asdum;
1047
if (asdum)
1048
{
1049
/* asdum->Release(); */
1050
}
1051
else
1052
SYNC_AUTOCOMMIT(conn);
1053
return SQL_SUCCESS;
1054
}
1055
if (CC_is_in_trans(conn))
1056
{
1057
CC_abort(conn);
1058
}
1059
if (!pDtc)
1060
{
1061
HRESULT res;
1062
1063
res = DtcGetTransactionManager(NULL, NULL, IID_ITransactionDispenser,
1064
0, 0, NULL, (void **) &pDtc);
1065
if (res != S_OK || !pDtc)
1066
{
1067
forcelog("TransactionManager get error %d\n", res);
1068
pDtc = NULL;
1069
}
1070
}
1071
return EnlistInDtc_1pipe(conn, (ITransaction *) pTra, pDtc);
1072
}
1073
1074
EXTERN_C RETCODE DtcOnDisconnect(ConnectionClass *conn)
1075
{
1076
mylog("DtcOnDisconnect\n");
1077
LIFELOCK_ACQUIRE;
1078
IAsyncPG *asdum = (IAsyncPG *) conn->asdum;
1079
if (asdum)
1080
{
1081
asdum->AddRef();
1082
LIFELOCK_RELEASE;
1083
asdum->ReleaseConnection();
1084
asdum->Release();
1085
}
1086
else
1087
LIFELOCK_RELEASE;
1088
return SQL_SUCCESS;
1089
}
1090
1091
EXTERN_C RETCODE DtcOnRelease(void)
1092
{
1093
AsyncThreads::CleanupThreads(2000);
1094
return SQL_SUCCESS;
1095
}
1096
1097
1098
#endif /* _HANDLE_ENLIST_IN_DTC_ */
1
/* Module: msdtc_enlist.cpp
2
*
3
* Description:
4
* This module contains routines related to
5
* the enlistment in MSDTC.
6
*
7
*-------
8
*/
9
10
#ifdef _HANDLE_ENLIST_IN_DTC_
11
12
#undef _MEMORY_DEBUG_
13
#ifndef _WIN32_WINNT
14
#define _WIN32_WINNT 0x0400
15
#endif /* _WIN32_WINNT */
16
17
#define WIN32_LEAN_AND_MEAN
18
#include <oleTx2xa.h>
19
#include <XOLEHLP.h>
20
/*#include <Txdtc.h>*/
21
#define _PGDTC_FUNCS_IMPORT_
22
#include "connexp.h"
23
24
/*#define _SLEEP_FOR_TEST_*/
25
#include <stdio.h>
26
#include <string.h>
27
#include <ctype.h>
28
#include <process.h>
29
#include <map>
30
#ifndef WIN32
31
#include <errno.h>
32
#endif /* WIN32 */
33
34
#include <sql.h>
35
#define _MYLOG_FUNCS_IMPORT_
36
#include "mylog.h"
37
#include "pgenlist.h"
38
39
#ifdef WIN32
40
#ifndef snprintf
41
#define snprintf _snprintf
42
#endif /* snprintf */
43
#endif /* WIN32 */
44
45
/* Define a type for defining a constant string expression */
46
#ifndef CSTR
47
#define CSTR static const char * const
48
#endif /* CSTR */
49
50
EXTERN_C {
51
HINSTANCE s_hModule; /* Saved module handle. */
52
}
53
/* This is where the Driver Manager attaches to this Driver */
54
BOOL WINAPI
55
DllMain(HANDLE hInst, ULONG ul_reason_for_call, LPVOID lpReserved)
56
{
57
switch (ul_reason_for_call)
58
{
59
case DLL_PROCESS_ATTACH:
60
s_hModule = (HINSTANCE) hInst; /* Save for dialog boxes
61
*/
62
break;
63
}
64
return TRUE;
65
}
66
67
/*
68
* A comment About locks used in this module
69
*
70
* the locks should be acquired with stronger to weaker order.
71
*
72
* 1:ELOCK -- the strongest per IAsyncPG object lock
73
* When the *isolated* or *dtcconn* member of an IAsyncPG object
74
* is changed, this lock should be held.
75
* While an IAsyncPG object accesses a psqlodbc connection,
76
* this lock should be held.
77
*
78
* 2:[CONN_CS] -- per psqlodbc connection lock
79
* This lock would be held for a pretty long time while accessing
80
* the psqlodbc connection assigned to an IAsyncPG object. You
81
* can use the connecion safely by holding a ELOCK for the
82
* IAsyncPG object because the assignment is ensured to be
83
* fixed while the ELOCK is held.
84
*
85
* 3:LIFELOCK -- a global lock to ensure the lives of IAsyncPG objects
86
* While this lock is held, IAsyncPG objects would never die.
87
*
88
* 4:SLOCK -- the short term per IAsyncPG object lock
89
* When any member of an IAsyncPG object is changed, this lock
90
* should be held.
91
*/
92
93
// #define _LOCK_DEBUG_
94
static class INIT_CRIT
95
{
96
public:
97
CRITICAL_SECTION life_cs; /* for asdum member of ConnectionClass */
98
INIT_CRIT() {
99
InitializeCriticalSection(&life_cs);
100
}
101
~INIT_CRIT() {
102
DeleteCriticalSection(&life_cs);
103
}
104
} init_crit;
105
#ifdef _LOCK_DEBUG_
106
#define LIFELOCK_ACQUIRE (forcelog("LIFELOCK_ACQUIRE\n"), EnterCriticalSection(&init_crit.life_cs), forcelog("LIFELOCK ACQUIRED\n"))
107
#define LIFELOCK_RELEASE (forcelog("LIFELOCK_RELEASE\n"), LeaveCriticalSection(&init_crit.life_cs))
108
#else
109
#define LIFELOCK_ACQUIRE EnterCriticalSection(&init_crit.life_cs)
110
#define LIFELOCK_RELEASE LeaveCriticalSection(&init_crit.life_cs)
111
#endif
112
113
/*
114
* Some helper macros about connection handling.
115
*/
116
#define CONN_CS_ACQUIRE(conn) PgDtc_lock_cntrl((conn), TRUE, FALSE)
117
#define TRY_CONN_CS_ACQUIRE(conn) PgDtc_lock_cntrl((conn), TRUE, TRUE)
118
#define CONN_CS_RELEASE(conn) PgDtc_lock_cntrl((conn), FALSE, FALSE)
119
120
#define CONN_IS_IN_TRANS(conn) PgDtc_get_property((conn), inTrans)
121
122
123
static const char *XidToText(const XID &xid, char *rtext)
124
{
125
int glen = xid.gtrid_length, blen = xid.bqual_length;
126
int i, j;
127
128
for (i = 0, j = 0; i < glen; i++, j += 2)
129
sprintf(rtext + j, "%02x", (unsigned char) xid.data[i]);
130
strcat(rtext, "-"); j++;
131
for (; i < glen + blen; i++, j += 2)
132
sprintf(rtext + j, "%02x", (unsigned char) xid.data[i]);
133
return rtext;
134
}
135
136
static LONG g_cComponents = 0;
137
static LONG g_cServerLocks = 0;
138
139
//
140
// �ȉ���ITransactionResourceAsync�I�u�W�F�N�g�͔C�ӂ̃X���b�h����
141
// ���R�ɃA�N�Z�X�\�Ȃ悤�Ɏ�������B�eRequest�̌��ʂ�Ԃ����߂�
142
// �g�p����ITransactionEnlistmentAsync�C���^�[�t�F�C�X�����̂悤��
143
// ��������Ă���i�Ǝv����A���L�Q�Ɓj�̂ŌĂяo����COM�̃A�p�[
144
// �g�����g���ӎ�����(CoMarshalInterThreadInterfaceInStream/CoGetIn
145
// terfaceAndReleaseStream���g�p����j�K�v�͂Ȃ��B
146
// ����DLL���Ŏg�p����ITransactionResourceAsync��ITransactionEnlist
147
// mentAsync�̃C���^�[�t�F�C�X�|�C���^�[�͔C�ӂ̃X���b�h���璼�ڎg�p
148
// ���邱�Ƃ��ł���B
149
//
150
151
// OLE Transactions Standard
152
//
153
// OLE Transactions is the Microsoft interface standard for transaction
154
// management. Applications use OLE Transactions-compliant interfaces to
155
// initiate, commit, abort, and inquire about transactions. Resource
156
// managers use OLE Transactions-compliant interfaces to enlist in
157
// transactions, to propagate transactions to other resource managers,
158
// to propagate transactions from process to process or from system to
159
// system, and to participate in the two-phase commit protocol.
160
//
161
// The Microsoft DTC system implements most OLE Transactions-compliant
162
// objects, interfaces, and methods. Resource managers that wish to use
163
// OLE Transactions must implement some OLE Transactions-compliant objects,
164
// interfaces, and methods.
165
//
166
// The OLE Transactions specification is based on COM but it differs in the
167
// following respects:
168
//
169
// OLE Transactions objects cannot be created using the COM CoCreate APIs.
170
// References to OLE Transactions objects are always direct. Therefore,
171
// no proxies or stubs are created for inter-apartment, inter-process,
172
// or inter-node calls and OLE Transactions references cannot be marshaled
173
// using standard COM marshaling.
174
// All references to OLE Transactions objects and their sinks are completely
175
// free threaded and cannot rely upon COM concurrency control models.
176
// For example, you cannot pass a reference to an IResourceManagerSink
177
// interface on a single-threaded apartment and expect the callback to occur
178
// only on the same single-threaded apartment.
179
180
class IAsyncPG : public ITransactionResourceAsync
181
{
182
private:
183
IDtcToXaHelperSinglePipe *helper;
184
DWORD RMCookie;
185
void *dtcconn;
186
LONG refcnt;
187
CRITICAL_SECTION as_spin; // to make this object Both
188
CRITICAL_SECTION as_exec; // to make this object Both
189
XID xid;
190
bool isolated;
191
bool prepared;
192
bool done;
193
bool abort;
194
HANDLE eThread[3];
195
bool eFin[3];
196
bool requestAccepted;
197
HRESULT prepare_result;
198
HRESULT commit_result;
199
#ifdef _LOCK_DEBUG_
200
int spin_cnt;
201
int cs_cnt;
202
#endif /* _LOCK_DEBUG_ */
203
204
public:
205
enum {
206
PrepareExec = 0
207
,CommitExec
208
,AbortExec
209
};
210
211
ITransactionEnlistmentAsync *enlist;
212
213
HRESULT STDMETHODCALLTYPE QueryInterface(REFIID iid, void ** ppvObject);
214
ULONG STDMETHODCALLTYPE AddRef(void);
215
ULONG STDMETHODCALLTYPE Release(void);
216
217
HRESULT STDMETHODCALLTYPE PrepareRequest(BOOL fRetaining,
218
DWORD grfRM,
219
BOOL fWantMoniker,
220
BOOL fSinglePhase);
221
HRESULT STDMETHODCALLTYPE CommitRequest(DWORD grfRM, XACTUOW * pNewUOW);
222
HRESULT STDMETHODCALLTYPE AbortRequest(BOID * pboidReason,
223
BOOL fRetaining,
224
XACTUOW * pNewUOW);
225
HRESULT STDMETHODCALLTYPE TMDown(void);
226
227
IAsyncPG();
228
void SetHelper(IDtcToXaHelperSinglePipe *pHelper, DWORD dwRMCookie) {helper = pHelper; RMCookie = dwRMCookie;}
229
230
HRESULT RequestExec(DWORD type, HRESULT res);
231
HRESULT ReleaseConnection(void);
232
void SetConnection(void *sconn) {SLOCK_ACQUIRE(); dtcconn = sconn; SLOCK_RELEASE();}
233
void SetXid(const XID *ixid) {SLOCK_ACQUIRE(); xid = *ixid; SLOCK_RELEASE();}
234
void *separateXAConn(bool spinAcquired, bool continueConnection);
235
bool CloseThread(DWORD type);
236
private:
237
~IAsyncPG();
238
#ifdef _LOCK_DEBUG_
239
void SLOCK_ACQUIRE() {forcelog("SLOCK_ACQUIRE %d\n", spin_cnt); EnterCriticalSection(&as_spin); spin_cnt++;}
240
void SLOCK_RELEASE() {forcelog("SLOCK_RELEASE=%d\n", spin_cnt); LeaveCriticalSection(&as_spin); spin_cnt--;}
241
void ELOCK_ACQUIRE() {forcelog("%p->ELOCK_ACQUIRE\n", this); EnterCriticalSection(&as_exec); forcelog("ELOCK ACQUIRED\n");}
242
void ELOCK_RELEASE() {forcelog("ELOCK_RELEASE\n"); LeaveCriticalSection(&as_exec);}
243
#else
244
void SLOCK_ACQUIRE() {EnterCriticalSection(&as_spin);}
245
void SLOCK_RELEASE() {LeaveCriticalSection(&as_spin);}
246
void ELOCK_ACQUIRE() {EnterCriticalSection(&as_exec);}
247
void ELOCK_RELEASE() {LeaveCriticalSection(&as_exec);}
248
#endif /* _LOCK_DEBUG_ */
249
void *getLockedXAConn(void);
250
void *generateXAConn(bool spinAcquired);
251
void *isolateXAConn(bool spinAcquired, bool continueConnection);
252
void SetPrepareResult(HRESULT res) {SLOCK_ACQUIRE(); prepared = true; prepare_result = res; SLOCK_RELEASE();}
253
void SetDone(HRESULT);
254
void Wait_pThread(bool slock_hold);
255
void Wait_cThread(bool slock_hold, bool once);
256
};
257
258
259
IAsyncPG::IAsyncPG(void) : helper(NULL), RMCookie(0), enlist(NULL), dtcconn(NULL), refcnt(1), isolated(false), done(false), abort(false), prepared(false), requestAccepted(false)
260
{
261
InterlockedIncrement(&g_cComponents);
262
InitializeCriticalSection(&as_spin);
263
InitializeCriticalSection(&as_exec);
264
eThread[0] = eThread[1] = eThread[2] = NULL;
265
eFin[0] = eFin[1] = eFin[2] = false;
266
memset(&xid, 0, sizeof(xid));
267
#ifdef _LOCK_DEBUG_
268
spin_cnt = 0;
269
cs_cnt = 0;
270
#endif /* _LOCK_DEBUG_ */
271
}
272
273
//
274
// invoked from *delete*.
275
// When entered ELOCK -> LIFELOCK -> SLOCK are held
276
// and they are released.
277
//
278
IAsyncPG::~IAsyncPG(void)
279
{
280
void *fconn = NULL;
281
282
if (dtcconn)
283
{
284
if (isolated)
285
fconn = dtcconn;
286
PgDtc_set_async(dtcconn, NULL);
287
dtcconn = NULL;
288
}
289
SLOCK_RELEASE();
290
LIFELOCK_RELEASE;
291
if (fconn)
292
{
293
mylog("IAsyncPG Destructor is freeing the connection\n");
294
PgDtc_free_connect(fconn);
295
}
296
DeleteCriticalSection(&as_spin);
297
ELOCK_RELEASE();
298
DeleteCriticalSection(&as_exec);
299
InterlockedDecrement(&g_cComponents);
300
}
301
HRESULT STDMETHODCALLTYPE IAsyncPG::QueryInterface(REFIID riid, void ** ppvObject)
302
{
303
forcelog("%p QueryInterface called\n", this);
304
if (riid == IID_IUnknown || riid == IID_ITransactionResourceAsync)
305
{
306
*ppvObject = this;
307
AddRef();
308
return S_OK;
309
}
310
*ppvObject = NULL;
311
return E_NOINTERFACE;
312
}
313
//
314
// acquire/releases SLOCK.
315
//
316
ULONG STDMETHODCALLTYPE IAsyncPG::AddRef(void)
317
{
318
mylog("%p->AddRef called\n", this);
319
SLOCK_ACQUIRE();
320
refcnt++;
321
SLOCK_RELEASE();
322
return refcnt;
323
}
324
//
325
// acquire/releases [ELOCK -> LIFELOCK -> ] SLOCK.
326
//
327
ULONG STDMETHODCALLTYPE IAsyncPG::Release(void)
328
{
329
mylog("%p->Release called refcnt=%d\n", this, refcnt);
330
SLOCK_ACQUIRE();
331
refcnt--;
332
if (refcnt <= 0)
333
{
334
SLOCK_RELEASE();
335
ELOCK_ACQUIRE();
336
LIFELOCK_ACQUIRE;
337
SLOCK_ACQUIRE();
338
if (refcnt <=0)
339
{
340
mylog("delete %p\n", this);
341
delete this;
342
}
343
else
344
{
345
SLOCK_RELEASE();
346
LIFELOCK_RELEASE;
347
ELOCK_RELEASE();
348
}
349
}
350
else
351
SLOCK_RELEASE();
352
return refcnt;
353
}
354
355
//
356
// Acquire/release SLOCK.
357
//
358
void IAsyncPG::Wait_pThread(bool slock_hold)
359
{
360
mylog("Wait_pThread %d in\n", slock_hold);
361
HANDLE wThread;
362
int wait_idx = PrepareExec;
363
DWORD ret;
364
365
if (!slock_hold)
366
SLOCK_ACQUIRE();
367
while (NULL != (wThread = eThread[wait_idx]) && !eFin[wait_idx])
368
{
369
SLOCK_RELEASE();
370
ret = WaitForSingleObject(wThread, 2000);
371
SLOCK_ACQUIRE();
372
if (WAIT_TIMEOUT != ret)
373
eFin[wait_idx] = true;
374
}
375
if (!slock_hold)
376
SLOCK_RELEASE();
377
mylog("Wait_pThread out\n");
378
}
379
380
//
381
// Acquire/releases SLOCK.
382
//
383
void IAsyncPG::Wait_cThread(bool slock_hold, bool once)
384
{
385
HANDLE wThread;
386
int wait_idx;
387
DWORD ret;
388
389
mylog("Wait_cThread %d,%d in\n", slock_hold, once);
390
if (!slock_hold)
391
SLOCK_ACQUIRE();
392
if (NULL != eThread[CommitExec])
393
wait_idx = CommitExec;
394
else
395
wait_idx = AbortExec;
396
while (NULL != (wThread = eThread[wait_idx]) && !eFin[wait_idx])
397
{
398
SLOCK_RELEASE();
399
ret = WaitForSingleObject(wThread, 2000);
400
SLOCK_ACQUIRE();
401
if (WAIT_TIMEOUT != ret)
402
eFin[wait_idx] = true;
403
else if (once)
404
break;
405
}
406
if (!slock_hold)
407
SLOCK_RELEASE();
408
mylog("Wait_cThread out\n");
409
}
410
411
/* Processing Prepare/Commit Request */
412
typedef
413
struct RequestPara {
414
DWORD type;
415
LPVOID lpr;
416
HRESULT res;
417
} RequestPara;
418
419
//
420
// Acquire/releases LIFELOCK -> SLOCK.
421
// may acquire/release ELOCK.
422
//
423
void IAsyncPG::SetDone(HRESULT res)
424
{
425
LIFELOCK_ACQUIRE;
426
SLOCK_ACQUIRE();
427
done = true;
428
if (E_FAIL == res ||
429
E_UNEXPECTED == res)
430
abort = true;
431
requestAccepted = true;
432
commit_result = res;
433
if (dtcconn)
434
{
435
PgDtc_set_async(dtcconn, NULL);
436
if (isolated)
437
{
438
SLOCK_RELEASE();
439
LIFELOCK_RELEASE;
440
ELOCK_ACQUIRE();
441
if (dtcconn)
442
{
443
mylog("Freeing isolated connection=%p\n", dtcconn);
444
PgDtc_free_connect(dtcconn);
445
SetConnection(NULL);
446
}
447
ELOCK_RELEASE();
448
}
449
else
450
{
451
dtcconn = NULL;
452
SLOCK_RELEASE();
453
LIFELOCK_RELEASE;
454
}
455
}
456
else
457
{
458
SLOCK_RELEASE();
459
LIFELOCK_RELEASE;
460
}
461
}
462
463
//
464
// Acquire/releases [ELOCK -> LIFELOCK -> ] SLOCK.
465
//
466
void *IAsyncPG::generateXAConn(bool spinAcquired)
467
{
468
mylog("generateXAConn isolated=%d dtcconn=%p\n", isolated, dtcconn);
469
if (!spinAcquired)
470
SLOCK_ACQUIRE();
471
if (isolated || done)
472
{
473
SLOCK_RELEASE();
474
return dtcconn;
475
}
476
SLOCK_RELEASE();
477
ELOCK_ACQUIRE();
478
LIFELOCK_ACQUIRE;
479
SLOCK_ACQUIRE();
480
if (dtcconn && !isolated && !done && prepared)
481
{
482
void *sconn = dtcconn;
483
484
dtcconn = PgDtc_isolate(sconn, useAnotherRoom);
485
isolated = true;
486
SLOCK_RELEASE();
487
LIFELOCK_RELEASE;
488
// PgDtc_connect(dtcconn); may be called in getLockedXAConn
489
}
490
else
491
{
492
SLOCK_RELEASE();
493
LIFELOCK_RELEASE;
494
}
495
ELOCK_RELEASE();
496
return dtcconn;
497
}
498
499
//
500
// Acquire/releases [ELOCK -> LIFELOCK -> ] SLOCK.
501
//
502
void *IAsyncPG::isolateXAConn(bool spinAcquired, bool continueConnection)
503
{
504
void *sconn;
505
506
mylog("isolateXAConn isolated=%d dtcconn=%p\n", isolated, dtcconn);
507
if (!spinAcquired)
508
SLOCK_ACQUIRE();
509
if (isolated || done || NULL == dtcconn)
510
{
511
SLOCK_RELEASE();
512
return dtcconn;
513
}
514
SLOCK_RELEASE();
515
ELOCK_ACQUIRE();
516
LIFELOCK_ACQUIRE;
517
SLOCK_ACQUIRE();
518
if (isolated || done || NULL == dtcconn)
519
{
520
SLOCK_RELEASE();
521
LIFELOCK_RELEASE;
522
ELOCK_RELEASE();
523
return dtcconn;
524
}
525
sconn = dtcconn;
526
527
dtcconn = PgDtc_isolate(sconn, continueConnection ? 0 : disposingConnection);
528
529
isolated = true;
530
SLOCK_RELEASE();
531
LIFELOCK_RELEASE;
532
if (continueConnection)
533
{
534
PgDtc_connect(sconn);
535
}
536
ELOCK_RELEASE();
537
return dtcconn;
538
}
539
540
//
541
// Acquire/releases [ELOCK -> LIFELOCK -> ] SLOCK.
542
//
543
void *IAsyncPG::separateXAConn(bool spinAcquired, bool continueConnection)
544
{
545
mylog("%s isolated=%d dtcconn=%p\n", __FUNCTION__, isolated, dtcconn);
546
if (!spinAcquired)
547
SLOCK_ACQUIRE();
548
if (prepared)
549
return generateXAConn(true);
550
else
551
return isolateXAConn(true, continueConnection);
552
}
553
554
//
555
// [when entered]
556
// ELOCK is held.
557
//
558
// Acquire/releases SLOCK.
559
// Try to acquire CONN_CS also.
560
//
561
// [on exit]
562
// ELOCK is kept held.
563
// If the return connection != NULL
564
// the CONN_CS lock for the connection is held.
565
//
566
void *IAsyncPG::getLockedXAConn()
567
{
568
SLOCK_ACQUIRE();
569
while (!done && !isolated && NULL != dtcconn)
570
{
571
/*
572
* Note that COMMIT/ROLLBACK PREPARED command should be
573
* issued outside the transaction.
574
*/
575
if (!prepared || !CONN_IS_IN_TRANS(dtcconn))
576
{
577
if (TRY_CONN_CS_ACQUIRE(dtcconn))
578
{
579
if (prepared && CONN_IS_IN_TRANS(dtcconn))
580
{
581
CONN_CS_RELEASE(dtcconn);
582
}
583
else
584
break;
585
}
586
}
587
separateXAConn(true, true);
588
SLOCK_ACQUIRE(); // SLOCK was released by separateXAConn()
589
}
590
SLOCK_RELEASE();
591
if (isolated && NULL != dtcconn)
592
{
593
CONN_CS_ACQUIRE(dtcconn);
594
if (!PgDtc_get_property(dtcconn, connected))
595
PgDtc_connect(dtcconn);
596
}
597
return dtcconn;
598
}
599
600
//
601
// Acquire/release ELOCK -> SLOCK.
602
//
603
HRESULT IAsyncPG::RequestExec(DWORD type, HRESULT res)
604
{
605
HRESULT ret;
606
bool bReleaseEnlist = false;
607
void *econn;
608
char pgxid[258];
609
610
mylog("%p->RequestExec type=%d conn=%p\n", this, type, dtcconn);
611
XidToText(xid, pgxid);
612
#ifdef _SLEEP_FOR_TEST_
613
/*Sleep(2000);*/
614
#endif /* _SLEEP_FOR_TEST_ */
615
ELOCK_ACQUIRE();
616
switch (type)
617
{
618
case PrepareExec:
619
if (done || NULL == dtcconn)
620
{
621
res = E_UNEXPECTED;
622
break;
623
}
624
if (econn = getLockedXAConn(), NULL != econn)
625
{
626
PgDtc_set_property(econn, inprogress, (void *) 1);
627
if (E_FAIL == res)
628
PgDtc_one_phase_operation(econn, ABORT_GLOBAL_TRANSACTION);
629
else if (XACT_S_SINGLEPHASE == res)
630
{
631
if (!PgDtc_one_phase_operation(econn, ONE_PHASE_COMMIT))
632
res = E_FAIL;
633
}
634
else
635
{
636
if (!PgDtc_two_phase_operation(econn, PREPARE_TRANSACTION, pgxid))
637
res = E_FAIL;
638
}
639
PgDtc_set_property(econn, inprogress, (void *) 0);
640
CONN_CS_RELEASE(econn);
641
}
642
if (S_OK != res)
643
{
644
SetDone(res);
645
bReleaseEnlist = true;
646
}
647
PgDtc_set_property(dtcconn, prepareRequested, (void *) 0);
648
ret = enlist->PrepareRequestDone(res, NULL, NULL);
649
SetPrepareResult(res);
650
break;
651
case CommitExec:
652
Wait_pThread(false);
653
if (E_FAIL != res)
654
{
655
econn = getLockedXAConn();
656
if (econn)
657
{
658
PgDtc_set_property(econn, inprogress, (void *) 1);
659
if (!PgDtc_two_phase_operation(econn, COMMIT_PREPARED, pgxid))
660
res = E_FAIL;
661
PgDtc_set_property(econn, inprogress, (void *) 0);
662
CONN_CS_RELEASE(econn);
663
}
664
}
665
SetDone(res);
666
ret = enlist->CommitRequestDone(res);
667
bReleaseEnlist = true;
668
break;
669
case AbortExec:
670
Wait_pThread(false);
671
if (prepared && !done)
672
{
673
econn = getLockedXAConn();
674
if (econn)
675
{
676
PgDtc_set_property(econn, inprogress, (void *) 1);
677
if (!PgDtc_two_phase_operation(econn, ROLLBACK_PREPARED, pgxid))
678
res = E_FAIL;
679
PgDtc_set_property(econn, inprogress, (void *) 0);
680
CONN_CS_RELEASE(econn);
681
}
682
}
683
SetDone(res);
684
ret = enlist->AbortRequestDone(res);
685
bReleaseEnlist = true;
686
break;
687
default:
688
ret = -1;
689
}
690
if (bReleaseEnlist)
691
{
692
helper->ReleaseRMCookie(RMCookie, TRUE);
693
enlist->Release();
694
}
695
ELOCK_RELEASE();
696
mylog("%p->Done ret=%d\n", this, ret);
697
return ret;
698
}
699
700
//
701
// Acquire/releses SLOCK
702
// or [ELOCK -> LIFELOCK -> ] SLOCK.
703
//
704
HRESULT IAsyncPG::ReleaseConnection(void)
705
{
706
mylog("%p->ReleaseConnection\n", this);
707
708
SLOCK_ACQUIRE();
709
if (isolated || NULL == dtcconn)
710
{
711
SLOCK_RELEASE();
712
return SQL_SUCCESS;
713
}
714
Wait_pThread(true);
715
if (NULL != eThread[CommitExec] || NULL != eThread[AbortExec] || requestAccepted)
716
{
717
if (!done)
718
Wait_cThread(true, true);
719
}
720
if (!isolated && !done && dtcconn && PgDtc_get_property(dtcconn, connected))
721
{
722
isolateXAConn(true, false);
723
}
724
else
725
SLOCK_RELEASE();
726
mylog("%p->ReleaseConnection exit\n", this);
727
return SQL_SUCCESS;
728
}
729
730
EXTERN_C static unsigned WINAPI DtcRequestExec(LPVOID para);
731
EXTERN_C static void __cdecl ClosePrepareThread(LPVOID para);
732
EXTERN_C static void __cdecl CloseCommitThread(LPVOID para);
733
EXTERN_C static void __cdecl CloseAbortThread(LPVOID para);
734
735
//
736
// Acquire/release [ELOCK -> ] SLOCK.
737
//
738
HRESULT STDMETHODCALLTYPE IAsyncPG::PrepareRequest(BOOL fRetaining, DWORD grfRM,
739
BOOL fWantMoniker, BOOL fSinglePhase)
740
{
741
HRESULT ret, res;
742
RequestPara *reqp;
743
const DWORD reqtype = PrepareExec;
744
745
mylog("%p PrepareRequest called grhRM=%d enl=%p\n", this, grfRM, enlist);
746
SLOCK_ACQUIRE();
747
if (dtcconn && 0 != PgDtc_get_property(dtcconn, errorNumber))
748
res = ret = E_FAIL;
749
else
750
{
751
ret = S_OK;
752
if (fSinglePhase)
753
{
754
res = XACT_S_SINGLEPHASE;
755
mylog("XACT is singlePhase\n");
756
}
757
else
758
res = S_OK;
759
}
760
SLOCK_RELEASE();
761
ELOCK_ACQUIRE();
762
#ifdef _SLEEP_FOR_TEST_
763
Sleep(2000);
764
#endif /* _SLEEP_FOR_TEST_ */
765
reqp = new RequestPara;
766
reqp->type = reqtype;
767
reqp->lpr = (LPVOID) this;
768
reqp->res = res;
769
#define DONT_CALL_RETURN_FROM_HERE ???
770
AddRef();
771
HANDLE hThread = (HANDLE) _beginthreadex(NULL, 0, DtcRequestExec, reqp, 0, NULL);
772
if (NULL == hThread)
773
{
774
delete(reqp);
775
ret = E_FAIL;
776
}
777
else
778
{
779
SLOCK_ACQUIRE();
780
eThread[reqtype] = hThread;
781
SLOCK_RELEASE();
782
/*
783
* We call here _beginthread not _beginthreadex
784
* so as not to call CloseHandle() to clean up
785
* the thread.
786
*/
787
_beginthread(ClosePrepareThread, 0, (void *) this);
788
}
789
ELOCK_RELEASE();
790
Release();
791
#undef return
792
return ret;
793
}
794
//
795
// Acquire/release [ELOCK -> ] SLOCK.
796
//
797
HRESULT STDMETHODCALLTYPE IAsyncPG::CommitRequest(DWORD grfRM, XACTUOW * pNewUOW)
798
{
799
HRESULT res = S_OK, ret = S_OK;
800
RequestPara *reqp;
801
const DWORD reqtype = CommitExec;
802
803
mylog("%p CommitRequest called grfRM=%d enl=%p\n", this, grfRM, enlist);
804
805
SLOCK_ACQUIRE();
806
if (!prepared || done)
807
ret = E_UNEXPECTED;
808
else if (S_OK != prepare_result)
809
ret = E_UNEXPECTED;
810
SLOCK_RELEASE();
811
if (S_OK != ret)
812
return ret;
813
#define DONT_CALL_RETURN_FROM_HERE ???
814
AddRef();
815
ELOCK_ACQUIRE();
816
#ifdef _SLEEP_FOR_TEST_
817
Sleep(1000);
818
#endif /* _SLEEP_FOR_TEST_ */
819
reqp = new RequestPara;
820
reqp->type = reqtype;
821
reqp->lpr = (LPVOID) this;
822
reqp->res = res;
823
enlist->AddRef();
824
HANDLE hThread = (HANDLE) _beginthreadex(NULL, 0, DtcRequestExec, reqp, 0, NULL);
825
if (NULL == hThread)
826
{
827
delete(reqp);
828
enlist->Release();
829
ret = E_FAIL;
830
}
831
else
832
{
833
SLOCK_ACQUIRE();
834
eThread[reqtype] = hThread;
835
SLOCK_RELEASE();
836
/*
837
* We call here _beginthread not _beginthreadex
838
* so as not to call CloseHandle() to clean up
839
* the thread.
840
*/
841
_beginthread(CloseCommitThread, 0, (void *) this);
842
}
843
mylog("CommitRequest ret=%d\n", ret);
844
requestAccepted = true;
845
ELOCK_RELEASE();
846
Release();
847
#undef return
848
return ret;
849
}
850
//
851
// Acquire/release [ELOCK -> ] SLOCK.
852
//
853
HRESULT STDMETHODCALLTYPE IAsyncPG::AbortRequest(BOID * pboidReason, BOOL fRetaining,
854
XACTUOW * pNewUOW)
855
{
856
HRESULT res = S_OK, ret = S_OK;
857
RequestPara *reqp;
858
const DWORD reqtype = AbortExec;
859
860
mylog("%p AbortRequest called\n", this);
861
SLOCK_ACQUIRE();
862
if (done)
863
ret = E_UNEXPECTED;
864
else if (prepared && S_OK != prepare_result)
865
ret = E_UNEXPECTED;
866
SLOCK_RELEASE();
867
if (S_OK != ret)
868
return ret;
869
#define return DONT_CALL_RETURN_FROM_HERE ???
870
AddRef();
871
ELOCK_ACQUIRE();
872
if (!prepared && dtcconn)
873
{
874
PgDtc_set_property(dtcconn, inprogress, (void *) 1);
875
PgDtc_one_phase_operation(dtcconn, ONE_PHASE_ROLLBACK);
876
PgDtc_set_property(dtcconn, inprogress, (void *) 0);
877
}
878
reqp = new RequestPara;
879
reqp->type = reqtype;
880
reqp->lpr = (LPVOID) this;
881
reqp->res = res;
882
enlist->AddRef();
883
HANDLE hThread = (HANDLE) _beginthreadex(NULL, 0, DtcRequestExec, reqp, 0, NULL);
884
if (NULL == hThread)
885
{
886
delete(reqp);
887
enlist->Release();
888
ret = E_FAIL;
889
}
890
else
891
{
892
SLOCK_ACQUIRE();
893
eThread[reqtype] = hThread;
894
SLOCK_RELEASE();
895
/*
896
* We call here _beginthread not _beginthreadex
897
* so as not to call CloseHandle() to clean up
898
* the thread.
899
*/
900
_beginthread(CloseAbortThread, 0, (void *) this);
901
}
902
mylog("AbortRequest ret=%d\n", ret);
903
requestAccepted = true;
904
ELOCK_RELEASE();
905
Release();
906
#undef return
907
return ret;
908
}
909
HRESULT STDMETHODCALLTYPE IAsyncPG::TMDown(void)
910
{
911
forcelog("%p TMDown called\n", this);
912
return S_OK;
913
}
914
915
bool IAsyncPG::CloseThread(DWORD type)
916
{
917
CSTR func = "CloseThread";
918
HANDLE th;
919
DWORD ret, excode = S_OK;
920
bool rls_async = false;
921
922
mylog("%s for %p thread=%d\n", func, this, eThread[type]);
923
if (th = eThread[type], NULL == th || eFin[type])
924
return false;
925
ret = WaitForSingleObject(th, INFINITE);
926
if (WAIT_OBJECT_0 == ret)
927
{
928
switch (type)
929
{
930
case IAsyncPG::AbortExec:
931
case IAsyncPG::CommitExec:
932
rls_async = true;
933
break;
934
default:
935
GetExitCodeThread(th, &excode);
936
if (S_OK != excode)
937
rls_async = true;
938
}
939
SLOCK_ACQUIRE();
940
eThread[type] = NULL;
941
eFin[type] = true;
942
SLOCK_RELEASE();
943
CloseHandle(th);
944
}
945
mylog("%s ret=%d\n", func, ret);
946
return rls_async;
947
}
948
949
EXTERN_C static void __cdecl ClosePrepareThread(LPVOID para)
950
{
951
CSTR func = "ClosePrepareThread";
952
IAsyncPG *async = (IAsyncPG *) para;
953
bool release;
954
955
mylog("%s for %p", func, async);
956
if (release = async->CloseThread(IAsyncPG::PrepareExec), release)
957
async->Release();
958
mylog("%s release=%d\n", func, release);
959
}
960
961
EXTERN_C static void __cdecl CloseCommitThread(LPVOID para)
962
{
963
CSTR func = "CloseCommitThread";
964
IAsyncPG *async = (IAsyncPG *) para;
965
bool release;
966
967
mylog("%s for %p", func, async);
968
if (release = async->CloseThread(IAsyncPG::CommitExec), release)
969
async->Release();
970
mylog("%s release=%d\n", func, release);
971
}
972
973
EXTERN_C static void __cdecl CloseAbortThread(LPVOID para)
974
{
975
CSTR func = "CloseAbortThread";
976
IAsyncPG *async = (IAsyncPG *) para;
977
bool release;
978
979
mylog("%s for %p", func, async);
980
if (release = async->CloseThread(IAsyncPG::AbortExec), release)
981
async->Release();
982
mylog("%s release=%d\n", func, release);
983
}
984
985
EXTERN_C static unsigned WINAPI DtcRequestExec(LPVOID para)
986
{
987
RequestPara *reqp = (RequestPara *) para;
988
DWORD type = reqp->type;
989
IAsyncPG *async = (IAsyncPG *) reqp->lpr;
990
HRESULT res = reqp->res, ret;
991
992
mylog("DtcRequestExec type=%d", reqp->type);
993
delete(reqp);
994
ret = async->RequestExec(type, res);
995
mylog(" Done ret=%d\n", ret);
996
return ret;
997
}
998
999
CSTR regKey = "SOFTWARE\\Microsoft\\MSDTC\\XADLL";
1000
1001
RETCODE static EnlistInDtc_1pipe(void *conn, ITransaction *pTra, ITransactionDispenser *pDtc)
1002
{
1003
CSTR func = "EnlistInDtc_1pipe";
1004
static IDtcToXaHelperSinglePipe *pHelper = NULL;
1005
ITransactionResourceAsync *pRes = NULL;
1006
IAsyncPG *asdum;
1007
HRESULT res;
1008
bool retry, errset;
1009
DWORD dwRMCookie;
1010
XID xid;
1011
1012
if (!pHelper)
1013
{
1014
res = pDtc->QueryInterface(IID_IDtcToXaHelperSinglePipe, (void **) &pHelper);
1015
if (res != S_OK || !pHelper)
1016
{
1017
forcelog("DtcToXaHelperSingelPipe get error %d\n", res);
1018
pHelper = NULL;
1019
return SQL_ERROR;
1020
}
1021
}
1022
res = (NULL != (asdum = new IAsyncPG)) ? S_OK : E_FAIL;
1023
if (S_OK != res)
1024
{
1025
mylog("CoCreateInstance error %d\n", res);
1026
return SQL_ERROR;
1027
}
1028
1029
/*mylog("dllname=%s dsn=%s\n", GetXaLibName(), conn->connInfo.dsn); res = 0;*/
1030
retry = false;
1031
errset = false;
1032
char dtcname[1024];
1033
PgDtc_create_connect_string(conn, dtcname, sizeof(dtcname));
1034
do {
1035
res = pHelper->XARMCreate(dtcname, (char *) GetXaLibName(), &dwRMCookie);
1036
if (S_OK == res)
1037
break;
1038
mylog("XARMCreate error code=%x\n", res);
1039
if (XACT_E_XA_TX_DISABLED == res)
1040
{
1041
PgDtc_set_error(conn, "XARMcreate error:Please enable XA transaction in MSDTC security configuration", func);
1042
errset = true;
1043
}
1044
else if (!retry)
1045
{
1046
LONG ret;
1047
HKEY sKey;
1048
DWORD rSize;
1049
1050
ret = ::RegOpenKeyEx(HKEY_LOCAL_MACHINE, regKey, 0, KEY_QUERY_VALUE | KEY_SET_VALUE, &sKey);
1051
if (ERROR_SUCCESS != ret)
1052
ret = ::RegCreateKeyEx(HKEY_LOCAL_MACHINE, regKey, 0, NULL, REG_OPTION_NON_VOLATILE, KEY_ALL_ACCESS, NULL, &sKey, NULL);
1053
if (ERROR_SUCCESS == ret)
1054
{
1055
switch (ret = ::RegQueryValueEx(sKey, "XADLL", NULL, NULL, NULL, &rSize))
1056
{
1057
case ERROR_SUCCESS:
1058
if (rSize > 0)
1059
break;
1060
default:
1061
ret = ::RegSetValueEx(sKey, GetXaLibName(), 0, REG_SZ,
1062
(CONST BYTE *) GetXaLibPath(), (DWORD) strlen(GetXaLibPath()) + 1);
1063
if (ERROR_SUCCESS == ret)
1064
{
1065
retry = true;
1066
continue; // retry
1067
}
1068
PgDtc_set_error(conn, "XARMCreate error:Please register HKLM\\SOFTWARE\\Microsoft\\MSDTC\\XADLL", func);
1069
break;
1070
}
1071
::RegCloseKey(sKey);
1072
}
1073
}
1074
if (!errset)
1075
PgDtc_set_error(conn, "MSDTC XARMCreate error", func);
1076
return SQL_ERROR;
1077
} while (1);
1078
res = pHelper->ConvertTridToXID((DWORD *) pTra, dwRMCookie, &xid);
1079
if (res != S_OK)
1080
{
1081
mylog("ConvertTridToXid error %d\n", res);
1082
return SQL_ERROR;
1083
}
1084
{
1085
char pgxid[258];
1086
XidToText(xid, pgxid);
1087
mylog("ConvertTridToXID -> %s\n", pgxid);
1088
}
1089
asdum->SetXid(&xid);
1090
/* Create an IAsyncPG instance by myself */
1091
/* DLLGetClassObject(GUID_IAsyncPG, IID_ITransactionResourceAsync, (void **) &asdum); */
1092
1093
asdum->SetHelper(pHelper, dwRMCookie);
1094
res = pHelper->EnlistWithRM(dwRMCookie, pTra, asdum, &asdum->enlist);
1095
if (res != S_OK)
1096
{
1097
mylog("EnlistWithRM error %d\n", res);
1098
pHelper->ReleaseRMCookie(dwRMCookie, TRUE);
1099
return SQL_ERROR;
1100
}
1101
1102
mylog("asdum=%p start transaction\n", asdum);
1103
asdum->SetConnection(conn);
1104
LIFELOCK_ACQUIRE;
1105
PgDtc_set_async(conn, asdum);
1106
LIFELOCK_RELEASE;
1107
1108
return SQL_SUCCESS;
1109
}
1110
1111
1112
EXTERN_C RETCODE
1113
IsolateDtcConn(void *conn, BOOL continueConnection)
1114
{
1115
IAsyncPG *async;
1116
1117
LIFELOCK_ACQUIRE;
1118
if (async = (IAsyncPG *) PgDtc_get_async(conn), NULL != async)
1119
{
1120
if (PgDtc_get_property(conn, idleInGlobalTransaction))
1121
{
1122
async->AddRef();
1123
LIFELOCK_RELEASE;
1124
async->separateXAConn(false, continueConnection ? true : false);
1125
async->Release();
1126
}
1127
else
1128
LIFELOCK_RELEASE;
1129
}
1130
else
1131
LIFELOCK_RELEASE;
1132
return SQL_SUCCESS;
1133
}
1134
1135
1136
EXTERN_C RETCODE EnlistInDtc(void *conn, void *pTra, int method)
1137
{
1138
static ITransactionDispenser *pDtc = NULL;
1139
RETCODE ret;
1140
1141
if (!pTra)
1142
{
1143
IAsyncPG *asdum = (IAsyncPG *) PgDtc_get_async(conn);
1144
PgDtc_set_property(conn, enlisted, (void *) 0);
1145
return SQL_SUCCESS;
1146
}
1147
if (CONN_IS_IN_TRANS(conn))
1148
{
1149
PgDtc_one_phase_operation(conn, SHUTDOWN_LOCAL_TRANSACTION);
1150
}
1151
if (!pDtc)
1152
{
1153
HRESULT res;
1154
1155
res = DtcGetTransactionManager(NULL, NULL, IID_ITransactionDispenser,
1156
0, 0, NULL, (void **) &pDtc);
1157
if (res != S_OK || !pDtc)
1158
{
1159
forcelog("TransactionManager get error %d\n", res);
1160
pDtc = NULL;
1161
}
1162
}
1163
ret = EnlistInDtc_1pipe(conn, (ITransaction *) pTra, pDtc);
1164
if (SQL_SUCCEEDED(ret))
1165
PgDtc_set_property(conn, enlisted, (void *) 1);
1166
return ret;
1167
}
1168
1169
EXTERN_C RETCODE DtcOnDisconnect(void *conn)
1170
{
1171
mylog("DtcOnDisconnect\n");
1172
LIFELOCK_ACQUIRE;
1173
IAsyncPG *asdum = (IAsyncPG *) PgDtc_get_async(conn);
1174
if (asdum)
1175
{
1176
asdum->AddRef();
1177
LIFELOCK_RELEASE;
1178
asdum->ReleaseConnection();
1179
asdum->Release();
1180
}
1181
else
1182
LIFELOCK_RELEASE;
1183
return SQL_SUCCESS;
1184
}
1185
1186
#endif /* _HANDLE_ENLIST_IN_DTC_ */
Loggerhead is a web-based interface for Breezy
Version: 2.0.1