2 Copyright (C) 2010 by Ronnie Sahlberg <ronniesahlberg@gmail.com>
4 This program is free software; you can redistribute it and/or modify
5 it under the terms of the GNU Lesser General Public License as published by
6 the Free Software Foundation; either version 2.1 of the License, or
7 (at your option) any later version.
9 This program is distributed in the hope that it will be useful,
10 but WITHOUT ANY WARRANTY; without even the implied warranty of
11 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 GNU Lesser General Public License for more details.
14 You should have received a copy of the GNU Lesser General Public License
15 along with this program; if not, see <http://www.gnu.org/licenses/>.
27 #include <arpa/inet.h>
28 #include <sys/ioctl.h>
30 #include "libnfs-raw.h"
31 #include "libnfs-private.h"
34 static void set_nonblocking(int fd
)
37 v
= fcntl(fd
, F_GETFL
, 0);
38 fcntl(fd
, F_SETFL
, v
| O_NONBLOCK
);
41 int rpc_get_fd(struct rpc_context
*rpc
)
46 int rpc_which_events(struct rpc_context
*rpc
)
48 int events
= rpc
->is_connected
? POLLIN
: POLLOUT
;
56 static int rpc_write_to_socket(struct rpc_context
*rpc
)
61 printf("trying to write to socket for NULL context\n");
65 printf("trying to write but not connected\n");
69 while (rpc
->outqueue
!= NULL
) {
72 total
= rpc
->outqueue
->outdata
.size
;
74 count
= write(rpc
->fd
, rpc
->outqueue
->outdata
.data
+ rpc
->outqueue
->written
, total
- rpc
->outqueue
->written
);
76 if (errno
== EAGAIN
|| errno
== EWOULDBLOCK
) {
77 printf("socket would block, return from write to socket\n");
80 printf("Error when writing to socket :%s(%d)\n", strerror(errno
), errno
);
84 rpc
->outqueue
->written
+= count
;
85 if (rpc
->outqueue
->written
== total
) {
86 struct rpc_pdu
*pdu
= rpc
->outqueue
;
88 SLIST_REMOVE(&rpc
->outqueue
, pdu
);
89 SLIST_ADD_END(&rpc
->waitpdu
, pdu
);
95 static int rpc_read_from_socket(struct rpc_context
*rpc
)
102 if (ioctl(rpc
->fd
, FIONREAD
, &available
) != 0) {
103 rpc_set_error(rpc
, "Ioctl FIONREAD returned error : %d. Closing socket.", errno
);
106 if (available
== 0) {
107 rpc_set_error(rpc
, "Socket has been closed");
110 size
= rpc
->insize
- rpc
->inpos
+ available
;
113 rpc_set_error(rpc
, "Out of memory: failed to allocate %d bytes for input buffer. Closing socket.", size
);
116 if (rpc
->insize
> rpc
->inpos
) {
117 memcpy(buf
, rpc
->inbuf
+ rpc
->inpos
, rpc
->insize
- rpc
->inpos
);
118 rpc
->insize
-= rpc
->inpos
;
122 count
= read(rpc
->fd
, buf
+ rpc
->insize
, available
);
124 if (errno
== EINTR
) {
129 rpc_set_error(rpc
, "Read from socket failed, errno:%d. Closing socket.", errno
);
135 if (rpc
->inbuf
!= NULL
) {
138 rpc
->inbuf
= (char *)buf
;
139 rpc
->insize
+= count
;
142 if (rpc
->insize
- rpc
->inpos
< 4) {
145 count
= rpc_get_pdu_size(rpc
->inbuf
+ rpc
->inpos
);
146 if (rpc
->insize
+ rpc
->inpos
< count
) {
149 if (rpc_process_pdu(rpc
, rpc
->inbuf
+ rpc
->inpos
, count
) != 0) {
150 rpc_set_error(rpc
, "Invalid/garbage pdu received from server. Closing socket");
154 if (rpc
->inpos
== rpc
->insize
) {
166 int rpc_service(struct rpc_context
*rpc
, int revents
)
168 if (revents
& POLLERR
) {
170 socklen_t err_size
= sizeof(err
);
172 if (getsockopt(rpc
->fd
, SOL_SOCKET
, SO_ERROR
,
173 &err
, &err_size
) != 0 || err
!= 0) {
177 rpc_set_error(rpc
, "rpc_service: socket error "
181 rpc_set_error(rpc
, "rpc_service: POLLERR, "
182 "Unknown socket error.");
184 rpc
->connect_cb(rpc
, RPC_STATUS_ERROR
, rpc
->error_string
, rpc
->connect_data
);
187 if (revents
& POLLHUP
) {
188 printf("rpc_service: POLLHUP, socket error\n");
189 rpc_set_error(rpc
, "Socket failed with POLLHUP");
190 rpc
->connect_cb(rpc
, RPC_STATUS_ERROR
, rpc
->error_string
, rpc
->connect_data
);
194 if (rpc
->is_connected
== 0 && rpc
->fd
!= -1 && revents
&POLLOUT
) {
196 socklen_t err_size
= sizeof(err
);
198 if (getsockopt(rpc
->fd
, SOL_SOCKET
, SO_ERROR
,
199 &err
, &err_size
) != 0 || err
!= 0) {
203 rpc_set_error(rpc
, "rpc_service: socket error "
204 "%s(%d) while connecting.",
206 rpc
->connect_cb(rpc
, RPC_STATUS_ERROR
,
207 NULL
, rpc
->connect_data
);
211 rpc
->is_connected
= 1;
212 rpc
->connect_cb(rpc
, RPC_STATUS_SUCCESS
, NULL
, rpc
->connect_data
);
216 if (revents
& POLLOUT
&& rpc
->outqueue
!= NULL
) {
217 if (rpc_write_to_socket(rpc
) != 0) {
218 printf("write to socket failed\n");
223 if (revents
& POLLIN
) {
224 if (rpc_read_from_socket(rpc
) != 0) {
225 rpc_disconnect(rpc
, rpc_get_error(rpc
));
234 int rpc_connect_async(struct rpc_context
*rpc
, const char *server
, int port
, rpc_cb cb
, void *private_data
)
236 struct sockaddr_storage s
;
237 struct sockaddr_in
*sin
= (struct sockaddr_in
*)&s
;
241 rpc_set_error(rpc
, "Trying to connect while already connected");
242 printf("%s\n", rpc
->error_string
);
246 sin
->sin_family
= AF_INET
;
247 sin
->sin_port
= htons(port
);
248 if (inet_pton(AF_INET
, server
, &sin
->sin_addr
) != 1) {
249 rpc_set_error(rpc
, "Not a valid server ip address");
250 printf("%s\n", rpc
->error_string
);
254 switch (s
.ss_family
) {
256 socksize
= sizeof(struct sockaddr_in
);
257 #ifdef HAVE_SOCK_SIN_LEN
258 sin
->sin_len
= socksize
;
260 rpc
->fd
= socket(AF_INET
, SOCK_STREAM
, 0);
265 rpc_set_error(rpc
, "Failed to open socket");
266 printf("%s\n", rpc
->error_string
);
270 rpc
->connect_cb
= cb
;
271 rpc
->connect_data
= private_data
;
273 set_nonblocking(rpc
->fd
);
275 if (connect(rpc
->fd
, (struct sockaddr
*)&s
, socksize
) != 0 && errno
!= EINPROGRESS
) {
276 rpc_set_error(rpc
, "connect() to server failed");
277 printf("%s\n", rpc
->error_string
);
284 int rpc_disconnect(struct rpc_context
*rpc
, char *error
)
291 rpc
->is_connected
= 0;
293 rpc_error_all_pdus(rpc
, error
);