Skip to content

Commit ebb4958

Browse files
knizhnikknizhnik
committed
Add ddd.c to dtmd
1 parent 3f42006 commit ebb4958

File tree

10 files changed

+65
-54
lines changed

10 files changed

+65
-54
lines changed

‎contrib/pg_dtm/docs/DTM.odp

537 KB
Binary file not shown.

‎contrib/pg_dtm/docs/dtm.pdf

-719 KB
Binary file not shown.

‎contrib/pg_dtm/dtmd/Makefile

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,11 +15,11 @@ all: bin/dtmd
1515
@echo Done.
1616
@echo Feel free to run the tests with \'make check\'.
1717

18-
bin/dtmd: obj/server.o obj/main.o obj/clog.o obj/clogfile.o obj/util.o obj/transaction.o obj/snapshot.o | bindir objdir
18+
bin/dtmd: obj/server.o obj/main.o obj/clog.o obj/clogfile.o obj/util.o obj/transaction.o obj/snapshot.o obj/ddd.o | bindir objdir
1919
$(CC) -o bin/dtmd $(CFLAGS) \
2020
obj/server.o obj/main.o \
2121
obj/clog.o obj/clogfile.o obj/util.o obj/transaction.o \
22-
obj/snapshot.o \
22+
obj/snapshot.o obj/ddd.o \
2323
$(SOCKHUB_LDFLAGS)
2424

2525
obj/server.o: src/server.c | objdir

‎contrib/pg_dtm/dtmd/include/transaction.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ typedef struct Transaction {
3434
void *listeners[CHAR_TO_INDEX('z')]; // we are going to use 'a' to 'z' for indexing
3535
} Transaction;
3636

37-
static inline void l2_list_is_empty(L2List* elem)
37+
static inline bool l2_list_is_empty(L2List* elem)
3838
{
3939
return elem->next == elem;
4040
}

‎contrib/pg_dtm/dtmd/src/ddd.c

Lines changed: 53 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -1,39 +1,19 @@
1-
#include "transaction.h"
1+
#include <stddef.h>
2+
#include <stdlib.h>
3+
#include <string.h>
4+
#include "ddd.h"
25

3-
typedef struct Instance {
4-
struct Edge* edges; /* local subgraph */
5-
} Instance;
6-
7-
typedef struct Edge {
8-
L2List node; /* node of list of outgoing eedges */
9-
struct Edge* next; /* list of edges of local subgraph */
10-
struct Vertex* dst;
11-
struct Vertex* src;
12-
} Edge;
13-
14-
typedef struct Vertex
15-
{
16-
L2List outgoingEdges;
17-
xid_t xid;
18-
int nIncomingEdges;
19-
bool visited;
20-
} Vertex;
21-
22-
typedef struct Graph
23-
{
24-
Vertex* hashtable[MAX_TRANSACTIONS];
25-
Edge* freeEdges;
26-
Vertex* freeVertexes;
27-
} Graph;
6+
static bool recursiveTraverseGraph(Vertex* root, Vertex* v, int marker);
287

298
void initGraph(Graph* graph)
309
{
31-
memset(graph->hashtable, 0. sizeof(graph->hashtable));
10+
memset(graph->hashtable, 0, sizeof(graph->hashtable));
3211
graph->freeEdges = NULL;
3312
graph->freeVertexes = NULL;
13+
graph->marker = 0;
3414
}
3515

36-
Edge* newEdge(Graph* graph)
16+
static inline Edge* newEdge(Graph* graph)
3717
{
3818
Edge* edge = graph->freeEdges;
3919
if (edge == NULL) {
@@ -44,44 +24,51 @@ Edge* newEdge(Graph* graph)
4424
return edge;
4525
}
4626

47-
void freeVertex(Graph* graph, Vertex* vertex)
27+
static inline void freeVertex(Graph* graph, Vertex* vertex)
4828
{
49-
vertex->node.next = (L2List*)graph->freeVertexes;
29+
int h = vertex->xid % MAX_TRANSACTIONS;
30+
Vertex** vpp = &graph->hashtable[h];
31+
while (*vpp != vertex) {
32+
vpp = &(*vpp)->next;
33+
}
34+
*vpp = vertex->next;
35+
vertex->next = graph->freeVertexes;
5036
graph->freeVertexes = vertex;
37+
5138
}
5239

53-
void freeEdge(Graph* graph, Edge* edge)
40+
static inline void freeEdge(Graph* graph, Edge* edge)
5441
{
5542
edge->next = graph->freeEdges;
5643
graph->freeEdges = edge;
5744
}
5845

59-
Vertex* newVertex(Graph* graph)
46+
static inline Vertex* newVertex(Graph* graph)
6047
{
6148
Vertex* v = graph->freeVertexes;
6249
if (v == NULL) {
6350
v = (Vertex*)malloc(sizeof(Vertex));
6451
} else {
65-
graph->freeVertexes = (Vertex*)v.node.next;
52+
graph->freeVertexes = v->next;
6653
}
6754
return v;
6855
}
6956

70-
Vertex* findVertex(Graph* graph, xid_t xid)
57+
static inline Vertex* findVertex(Graph* graph, xid_t xid)
7158
{
72-
xid_t h = xid;
59+
xid_t h = xid % MAX_TRANSACTIONS;
7360
Vertex* v;
74-
while ((v = graph->hashtable[h % MAX_TRANSACTIONS]) != NULL) {
61+
for (v = graph->hashtable[h]; v != NULL; v = v->next) {
7562
if (v->xid == xid) {
7663
return v;
7764
}
78-
h += 1;
7965
}
8066
v = newVertex(graph);
81-
l2_list_init(v->outgoingEdges);
67+
l2_list_init(&v->outgoingEdges);
8268
v->xid = xid;
8369
v->nIncomingEdges = 0;
84-
graph->hashtable[h % MAX_TRANSACTIONS] = v;
70+
v->next = graph->hashtable[h];
71+
graph->hashtable[h] = v;
8572
return v;
8673
}
8774

@@ -107,15 +94,38 @@ void addSubgraph(Instance* instance, Graph* graph, xid_t* xids, int n_xids)
10794
next = e->next;
10895
l2_list_unlink(&e->node);
10996
if (--e->dst->nIncomingEdges == 0 && l2_list_is_empty(&e->dst->outgoingEdges)) {
110-
freeVertex(e->dst);
97+
freeVertex(graph, e->dst);
11198
}
11299
if (e->src->nIncomingEdges == 0 && l2_list_is_empty(&e->src->outgoingEdges)) {
113-
freeVertex(e->src);
100+
freeVertex(graph, e->src);
114101
}
115-
freeEdge(e);
102+
freeEdge(graph, e);
116103
}
117104
}
118105

119-
bool findLoop(Graph* graph)
106+
static bool recursiveTraverseGraph(Vertex* root, Vertex* v, int marker)
120107
{
108+
L2List* l;
109+
Edge* e;
110+
v->visited = marker;
111+
for (l = v->outgoingEdges.next; l != &v->outgoingEdges; l = e->node.next) {
112+
e = (Edge*)l;
113+
if (e->dst == root) {
114+
return true;
115+
} else if (e->dst->visited != marker && recursiveTraverseGraph(root, e->dst, marker)) { /* loop */
116+
return true;
117+
}
118+
}
119+
return false;
120+
}
121+
122+
bool findLoop(Graph* graph, xid_t root)
123+
{
124+
Vertex* v;
125+
for (v = graph->hashtable[root % MAX_TRANSACTIONS]; v != NULL; v = v->next) {
126+
if (v->xid == root) {
127+
return recursiveTraverseGraph(v, v, ++graph->marker);
128+
}
129+
}
130+
return false;
121131
}

‎contrib/pg_dtm/libdtm.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -500,7 +500,7 @@ int DtmGlobalReserve(TransactionId xid, int nXids, TransactionId *first)
500500
return 0;
501501
}
502502

503-
bool DtmGlobalDetectDeadLock(void* data, int size)
503+
bool DtmGlobalDetectDeadLock(TransactionId xid, void* data, int size)
504504
{
505505
return false;
506506
}

‎contrib/pg_dtm/libdtm.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,6 @@ int DtmGlobalReserve(TransactionId xid, int nXids, TransactionId *first);
5959
* Once loop is detected in global resoruce graph, arbiter returns true. Otherwise false is returned.
6060
* Abiter should replace local part of resource graph if new graph is recevied from this cluster node (not backend).
6161
*/
62-
bool DtmGlobalDetectDeadLock(void* graph, int size);
62+
bool DtmGlobalDetectDeadLock(TransactionId xid, void* graph, int size);
6363

6464
#endif

‎contrib/pg_dtm/pg_dtm.c

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@ static TransactionId DtmGetNextXid(void);
7979
static TransactionId DtmGetNewTransactionId(bool isSubXact);
8080
static TransactionId DtmGetOldestXmin(Relation rel, bool ignoreVacuum);
8181
static TransactionId DtmGetGlobalTransactionId(void);
82-
static bool DtmDetectGlobalDeadLock(void);
82+
static bool DtmDetectGlobalDeadLock(PGPROC* proc);
8383

8484
static void DtmSerializeLock(PROCLOCK* lock, void* arg);
8585

@@ -1023,13 +1023,13 @@ static void DtmSerializeLock(PROCLOCK* proclock, void* arg)
10231023
}
10241024
}
10251025

1026-
bool DtmDetectGlobalDeadLock(void)
1026+
bool DtmDetectGlobalDeadLock(PGPROC* proc)
10271027
{
10281028
bool hasDeadlock;
10291029
ByteBuffer buf;
10301030
ByteBufferAlloc(&buf);
10311031
EnumerateLocks(DtmSerializeLock, &buf);
1032-
hasDeadlock = DtmGlobalDetectDeadLock(buf.data, buf.used);
1032+
hasDeadlock = DtmGlobalDetectDeadLock(proc->lxid, buf.data, buf.used);
10331033
ByteBufferFree(&buf);
10341034
return hasDeadlock;
10351035
}

‎src/backend/access/transam/xtm.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ TransactionId PgGetGlobalTransactionId(void)
2323
return InvalidTransactionId;
2424
}
2525

26-
bool PgDetectGlobalDeadLock()
26+
bool PgDetectGlobalDeadLock(PGPROC* proc)
2727
{
2828
return false;
2929
}

‎src/include/access/xtm.h

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
#ifndef XTM_H
1212
#define XTM_H
1313

14+
#include "storage/proc.h"
1415
#include "access/clog.h"
1516
#include "utils/snapmgr.h"
1617
#include "utils/relcache.h"
@@ -42,7 +43,7 @@ typedef struct
4243
bool (*IsInSnapshot)(TransactionId xid, Snapshot snapshot);
4344

4445
/* Detect distributed deadlock */
45-
bool (*DetectGlobalDeadLock)(void);
46+
bool (*DetectGlobalDeadLock)(PGPROC* proc);
4647
} TransactionManager;
4748

4849
/* Get pointer to transaction manager: actually returns content of TM variable */
@@ -68,6 +69,6 @@ extern TransactionId PgGetGlobalTransactionId(void);
6869

6970
extern TransactionId PgGetNewTransactionId(bool isSubXact);
7071

71-
extern bool PgDetectGlobalDeadLock(void);
72+
extern bool PgDetectGlobalDeadLock(PGPROC* proc);
7273

7374
#endif

0 commit comments

Comments
 (0)