1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
|
/* Copyright (C) 2003 MySQL AB
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; version 2 of the License.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA */
#ifndef TCP_TRANSPORTER_HPP
#define TCP_TRANSPORTER_HPP
#include "Transporter.hpp"
#include "SendBuffer.hpp"
#include <NdbTCP.h>
struct ReceiveBuffer {
Uint32 *startOfBuffer; // Pointer to start of the receive buffer
Uint32 *readPtr; // Pointer to start reading data
char *insertPtr; // Pointer to first position in the receiveBuffer
// in which to insert received data. Earlier
// received incomplete messages (slack) are
// copied into the first part of the receiveBuffer
Uint32 sizeOfData; // In bytes
Uint32 sizeOfBuffer;
ReceiveBuffer() {}
bool init(int bytes);
void destroy();
void clear();
void incompleteMessage();
};
class TCP_Transporter : public Transporter {
friend class TransporterRegistry;
private:
// Initialize member variables
TCP_Transporter(TransporterRegistry&,
int sendBufferSize, int maxReceiveSize,
const char *lHostName,
const char *rHostName,
int r_port,
bool isMgmConnection,
NodeId lHostId,
NodeId rHostId,
NodeId serverNodeId,
bool checksum, bool signalId,
Uint32 reportFreq = 4096);
// Disconnect, delete send buffers and receive buffer
virtual ~TCP_Transporter();
/**
* Allocate buffers for sending and receiving
*/
bool initTransporter();
Uint32 * getWritePtr(Uint32 lenBytes, Uint32 prio);
void updateWritePtr(Uint32 lenBytes, Uint32 prio);
bool hasDataToSend() const ;
/**
* Retrieves the contents of the send buffers and writes it on
* the external TCP/IP interface until the send buffers are empty
* and as long as write is possible.
*/
bool doSend();
/**
* It reads the external TCP/IP interface once
* and puts the data in the receiveBuffer
*/
int doReceive();
/**
* Returns socket (used for select)
*/
NDB_SOCKET_TYPE getSocket() const;
/**
* Get Receive Data
*
* Returns - no of bytes to read
* and set ptr
*/
virtual Uint32 getReceiveData(Uint32 ** ptr);
/**
* Update receive data ptr
*/
virtual void updateReceiveDataPtr(Uint32 bytesRead);
virtual Uint32 get_free_buffer() const;
inline bool hasReceiveData () const {
return receiveBuffer.sizeOfData > 0;
}
protected:
/**
* Setup client/server and perform connect/accept
* Is used both by clients and servers
* A client connects to the remote server
* A server accepts any new connections
*/
virtual bool connect_server_impl(NDB_SOCKET_TYPE sockfd);
virtual bool connect_client_impl(NDB_SOCKET_TYPE sockfd);
bool connect_common(NDB_SOCKET_TYPE sockfd);
/**
* Disconnects a TCP/IP node. Empty send and receivebuffer.
*/
virtual void disconnectImpl();
private:
/**
* Send buffers
*/
SendBuffer m_sendBuffer;
// Sending/Receiving socket used by both client and server
NDB_SOCKET_TYPE theSocket;
Uint32 maxReceiveSize;
/**
* Socket options
*/
int sockOptRcvBufSize;
int sockOptSndBufSize;
int sockOptNodelay;
int sockOptTcpMaxSeg;
void setSocketOptions();
static bool setSocketNonBlocking(NDB_SOCKET_TYPE aSocket);
bool sendIsPossible(struct timeval * timeout);
/**
* Statistics
*/
Uint32 reportFreq;
Uint32 receiveCount;
Uint64 receiveSize;
Uint32 sendCount;
Uint64 sendSize;
ReceiveBuffer receiveBuffer;
};
inline
NDB_SOCKET_TYPE
TCP_Transporter::getSocket() const {
return theSocket;
}
inline
Uint32
TCP_Transporter::getReceiveData(Uint32 ** ptr){
(* ptr) = receiveBuffer.readPtr;
return receiveBuffer.sizeOfData;
}
inline
void
TCP_Transporter::updateReceiveDataPtr(Uint32 bytesRead){
char * ptr = (char *)receiveBuffer.readPtr;
ptr += bytesRead;
receiveBuffer.readPtr = (Uint32*)ptr;
receiveBuffer.sizeOfData -= bytesRead;
receiveBuffer.incompleteMessage();
}
inline
bool
TCP_Transporter::hasDataToSend() const {
return m_sendBuffer.dataSize > 0;
}
inline
bool
ReceiveBuffer::init(int bytes){
#ifdef DEBUG_TRANSPORTER
ndbout << "Allocating " << bytes << " bytes as receivebuffer" << endl;
#endif
startOfBuffer = new Uint32[((bytes + 0) >> 2) + 1];
sizeOfBuffer = bytes + sizeof(Uint32);
clear();
return true;
}
inline
void
ReceiveBuffer::destroy(){
delete[] startOfBuffer;
sizeOfBuffer = 0;
startOfBuffer = 0;
clear();
}
inline
void
ReceiveBuffer::clear(){
readPtr = startOfBuffer;
insertPtr = (char *)startOfBuffer;
sizeOfData = 0;
}
inline
void
ReceiveBuffer::incompleteMessage() {
if(startOfBuffer != readPtr){
if(sizeOfData != 0)
memmove(startOfBuffer, readPtr, sizeOfData);
readPtr = startOfBuffer;
insertPtr = ((char *)startOfBuffer) + sizeOfData;
}
}
#endif // Define of TCP_Transporter_H
|
