@matheusbsbs / matheus-bagliano / commits / 419f37db4d
Fork 0

Add bidirectional_transfer_loop()

This helper function copies bidirectional stream of data between stdin/stdout and specified file descriptors. Signed-off-by: Ilari Liusvaara <ilari.liusvaara@elisanet.fi> Signed-off-by: Junio C Hamano <gitster@pobox.com>

Ilari Liusvaara committed Oct 12, 2010 at 19:39 UTC 419f37db4d1e2fe3c394c869c5375e545b3e364d
3 files changed +317
compat/mingw.h
+5
@@ -23,6 +23,9 @@ typedef int pid_t;
23 #define WEXITSTATUS(x) ((x) & 0xff)
24 #define WTERMSIG(x) SIGTERM
25
26 +#define EWOULDBLOCK EAGAIN
27 +#define SHUT_WR SD_SEND
28 +
29 #define SIGHUP 1
30 #define SIGQUIT 3
31 #define SIGKILL 9
@@ -50,6 +53,8 @@ struct pollfd {
53 };
54 #define POLLIN 1
55 #define POLLHUP 2
56 +#define POLLOUT 4
57 +#define POLLNVAL 8
58 #endif
59
60 typedef void (__cdecl *sig_handler_t)(int);
transport-helper.c
+311
@@ -862,3 +862,314 @@ int transport_helper_init(struct transport *transport, const char *name)
862 transport->smart_options = &(data->transport_options);
863 return 0;
864 }
865 +
866 +/*
867 + * Linux pipes can buffer 65536 bytes at once (and most platforms can
868 + * buffer less), so attempt reads and writes with up to that size.
869 + */
870 +#define BUFFERSIZE 65536
871 +/* This should be enough to hold debugging message. */
872 +#define PBUFFERSIZE 8192
873 +
874 +/* Print bidirectional transfer loop debug message. */
875 +static void transfer_debug(const char *fmt, ...)
876 +{
877 + va_list args;
878 + char msgbuf[PBUFFERSIZE];
879 + static int debug_enabled = -1;
880 +
881 + if (debug_enabled < 0)
882 + debug_enabled = getenv("GIT_TRANSLOOP_DEBUG") ? 1 : 0;
883 + if (!debug_enabled)
884 + return;
885 +
886 + va_start(args, fmt);
887 + vsnprintf(msgbuf, PBUFFERSIZE, fmt, args);
888 + va_end(args);
889 + fprintf(stderr, "Transfer loop debugging: %s\n", msgbuf);
890 +}
891 +
892 +/* Stream state: More data may be coming in this direction. */
893 +#define SSTATE_TRANSFERING 0
894 +/*
895 + * Stream state: No more data coming in this direction, flushing rest of
896 + * data.
897 + */
898 +#define SSTATE_FLUSHING 1
899 +/* Stream state: Transfer in this direction finished. */
900 +#define SSTATE_FINISHED 2
901 +
902 +#define STATE_NEEDS_READING(state) ((state) <= SSTATE_TRANSFERING)
903 +#define STATE_NEEDS_WRITING(state) ((state) <= SSTATE_FLUSHING)
904 +#define STATE_NEEDS_CLOSING(state) ((state) == SSTATE_FLUSHING)
905 +
906 +/* Unidirectional transfer. */
907 +struct unidirectional_transfer {
908 + /* Source */
909 + int src;
910 + /* Destination */
911 + int dest;
912 + /* Is source socket? */
913 + int src_is_sock;
914 + /* Is destination socket? */
915 + int dest_is_sock;
916 + /* Transfer state (TRANSFERING/FLUSHING/FINISHED) */
917 + int state;
918 + /* Buffer. */
919 + char buf[BUFFERSIZE];
920 + /* Buffer used. */
921 + size_t bufuse;
922 + /* Name of source. */
923 + const char *src_name;
924 + /* Name of destination. */
925 + const char *dest_name;
926 +};
927 +
928 +/* Closes the target (for writing) if transfer has finished. */
929 +static void udt_close_if_finished(struct unidirectional_transfer *t)
930 +{
931 + if (STATE_NEEDS_CLOSING(t->state) && !t->bufuse) {
932 + t->state = SSTATE_FINISHED;
933 + if (t->dest_is_sock)
934 + shutdown(t->dest, SHUT_WR);
935 + else
936 + close(t->dest);
937 + transfer_debug("Closed %s.", t->dest_name);
938 + }
939 +}
940 +
941 +/*
942 + * Tries to read read data from source into buffer. If buffer is full,
943 + * no data is read. Returns 0 on success, -1 on error.
944 + */
945 +static int udt_do_read(struct unidirectional_transfer *t)
946 +{
947 + ssize_t bytes;
948 +
949 + if (t->bufuse == BUFFERSIZE)
950 + return 0; /* No space for more. */
951 +
952 + transfer_debug("%s is readable", t->src_name);
953 + bytes = read(t->src, t->buf + t->bufuse, BUFFERSIZE - t->bufuse);
954 + if (bytes < 0 && errno != EWOULDBLOCK && errno != EAGAIN &&
955 + errno != EINTR) {
956 + error("read(%s) failed: %s", t->src_name, strerror(errno));
957 + return -1;
958 + } else if (bytes == 0) {
959 + transfer_debug("%s EOF (with %i bytes in buffer)",
960 + t->src_name, t->bufuse);
961 + t->state = SSTATE_FLUSHING;
962 + } else if (bytes > 0) {
963 + t->bufuse += bytes;
964 + transfer_debug("Read %i bytes from %s (buffer now at %i)",
965 + (int)bytes, t->src_name, (int)t->bufuse);
966 + }
967 + return 0;
968 +}
969 +
970 +/* Tries to write data from buffer into destination. If buffer is empty,
971 + * no data is written. Returns 0 on success, -1 on error.
972 + */
973 +static int udt_do_write(struct unidirectional_transfer *t)
974 +{
975 + size_t bytes;
976 +
977 + if (t->bufuse == 0)
978 + return 0; /* Nothing to write. */
979 +
980 + transfer_debug("%s is writable", t->dest_name);
981 + bytes = write(t->dest, t->buf, t->bufuse);
982 + if (bytes < 0 && errno != EWOULDBLOCK && errno != EAGAIN &&
983 + errno != EINTR) {
984 + error("write(%s) failed: %s", t->dest_name, strerror(errno));
985 + return -1;
986 + } else if (bytes > 0) {
987 + t->bufuse -= bytes;
988 + if (t->bufuse)
989 + memmove(t->buf, t->buf + bytes, t->bufuse);
990 + transfer_debug("Wrote %i bytes to %s (buffer now at %i)",
991 + (int)bytes, t->dest_name, (int)t->bufuse);
992 + }
993 + return 0;
994 +}
995 +
996 +
997 +/* State of bidirectional transfer loop. */
998 +struct bidirectional_transfer_state {
999 + /* Direction from program to git. */
1000 + struct unidirectional_transfer ptg;
1001 + /* Direction from git to program. */
1002 + struct unidirectional_transfer gtp;
1003 +};
1004 +
1005 +static void *udt_copy_task_routine(void *udt)
1006 +{
1007 + struct unidirectional_transfer *t = (struct unidirectional_transfer *)udt;
1008 + while (t->state != SSTATE_FINISHED) {
1009 + if (STATE_NEEDS_READING(t->state))
1010 + if (udt_do_read(t))
1011 + return NULL;
1012 + if (STATE_NEEDS_WRITING(t->state))
1013 + if (udt_do_write(t))
1014 + return NULL;
1015 + if (STATE_NEEDS_CLOSING(t->state))
1016 + udt_close_if_finished(t);
1017 + }
1018 + return udt; /* Just some non-NULL value. */
1019 +}
1020 +
1021 +#ifndef NO_PTHREADS
1022 +
1023 +/*
1024 + * Join thread, with apporiate errors on failure. Name is name for the
1025 + * thread (for error messages). Returns 0 on success, 1 on failure.
1026 + */
1027 +static int tloop_join(pthread_t thread, const char *name)
1028 +{
1029 + int err;
1030 + void *tret;
1031 + err = pthread_join(thread, &tret);
1032 + if (!tret) {
1033 + error("%s thread failed", name);
1034 + return 1;
1035 + }
1036 + if (err) {
1037 + error("%s thread failed to join: %s", name, strerror(err));
1038 + return 1;
1039 + }
1040 + return 0;
1041 +}
1042 +
1043 +/*
1044 + * Spawn the transfer tasks and then wait for them. Returns 0 on success,
1045 + * -1 on failure.
1046 + */
1047 +static int tloop_spawnwait_tasks(struct bidirectional_transfer_state *s)
1048 +{
1049 + pthread_t gtp_thread;
1050 + pthread_t ptg_thread;
1051 + int err;
1052 + int ret = 0;
1053 + err = pthread_create(&gtp_thread, NULL, udt_copy_task_routine,
1054 + &s->gtp);
1055 + if (err)
1056 + die("Can't start thread for copying data: %s", strerror(err));
1057 + err = pthread_create(&ptg_thread, NULL, udt_copy_task_routine,
1058 + &s->ptg);
1059 + if (err)
1060 + die("Can't start thread for copying data: %s", strerror(err));
1061 +
1062 + ret |= tloop_join(gtp_thread, "Git to program copy");
1063 + ret |= tloop_join(ptg_thread, "Program to git copy");
1064 + return ret;
1065 +}
1066 +#else
1067 +
1068 +/* Close the source and target (for writing) for transfer. */
1069 +static void udt_kill_transfer(struct unidirectional_transfer *t)
1070 +{
1071 + t->state = SSTATE_FINISHED;
1072 + /*
1073 + * Socket read end left open isn't a disaster if nobody
1074 + * attempts to read from it (mingw compat headers do not
1075 + * have SHUT_RD)...
1076 + *
1077 + * We can't fully close the socket since otherwise gtp
1078 + * task would first close the socket it sends data to
1079 + * while closing the ptg file descriptors.
1080 + */
1081 + if (!t->src_is_sock)
1082 + close(t->src);
1083 + if (t->dest_is_sock)
1084 + shutdown(t->dest, SHUT_WR);
1085 + else
1086 + close(t->dest);
1087 +}
1088 +
1089 +/*
1090 + * Join process, with apporiate errors on failure. Name is name for the
1091 + * process (for error messages). Returns 0 on success, 1 on failure.
1092 + */
1093 +static int tloop_join(pid_t pid, const char *name)
1094 +{
1095 + int tret;
1096 + if (waitpid(pid, &tret, 0) < 0) {
1097 + error("%s process failed to wait: %s", name, strerror(errno));
1098 + return 1;
1099 + }
1100 + if (!WIFEXITED(tret) || WEXITSTATUS(tret)) {
1101 + error("%s process failed", name);
1102 + return 1;
1103 + }
1104 + return 0;
1105 +}
1106 +
1107 +/*
1108 + * Spawn the transfer tasks and then wait for them. Returns 0 on success,
1109 + * -1 on failure.
1110 + */
1111 +static int tloop_spawnwait_tasks(struct bidirectional_transfer_state *s)
1112 +{
1113 + pid_t pid1, pid2;
1114 + int ret = 0;
1115 +
1116 + /* Fork thread #1: git to program. */
1117 + pid1 = fork();
1118 + if (pid1 < 0)
1119 + die_errno("Can't start thread for copying data");
1120 + else if (pid1 == 0) {
1121 + udt_kill_transfer(&s->ptg);
1122 + exit(udt_copy_task_routine(&s->gtp) ? 0 : 1);
1123 + }
1124 +
1125 + /* Fork thread #2: program to git. */
1126 + pid2 = fork();
1127 + if (pid2 < 0)
1128 + die_errno("Can't start thread for copying data");
1129 + else if (pid2 == 0) {
1130 + udt_kill_transfer(&s->gtp);
1131 + exit(udt_copy_task_routine(&s->ptg) ? 0 : 1);
1132 + }
1133 +
1134 + /*
1135 + * Close both streams in parent as to not interfere with
1136 + * end of file detection and wait for both tasks to finish.
1137 + */
1138 + udt_kill_transfer(&s->gtp);
1139 + udt_kill_transfer(&s->ptg);
1140 + ret |= tloop_join(pid1, "Git to program copy");
1141 + ret |= tloop_join(pid2, "Program to git copy");
1142 + return ret;
1143 +}
1144 +#endif
1145 +
1146 +/*
1147 + * Copies data from stdin to output and from input to stdout simultaneously.
1148 + * Additionally filtering through given filter. If filter is NULL, uses
1149 + * identity filter.
1150 + */
1151 +int bidirectional_transfer_loop(int input, int output)
1152 +{
1153 + struct bidirectional_transfer_state state;
1154 +
1155 + /* Fill the state fields. */
1156 + state.ptg.src = input;
1157 + state.ptg.dest = 1;
1158 + state.ptg.src_is_sock = (input == output);
1159 + state.ptg.dest_is_sock = 0;
1160 + state.ptg.state = SSTATE_TRANSFERING;
1161 + state.ptg.bufuse = 0;
1162 + state.ptg.src_name = "remote input";
1163 + state.ptg.dest_name = "stdout";
1164 +
1165 + state.gtp.src = 0;
1166 + state.gtp.dest = output;
1167 + state.gtp.src_is_sock = 0;
1168 + state.gtp.dest_is_sock = (input == output);
1169 + state.gtp.state = SSTATE_TRANSFERING;
1170 + state.gtp.bufuse = 0;
1171 + state.gtp.src_name = "stdin";
1172 + state.gtp.dest_name = "remote output";
1173 +
1174 + return tloop_spawnwait_tasks(&state);
1175 +}
transport.h
+1
@@ -154,6 +154,7 @@ int transport_connect(struct transport *transport, const char *name,
154
155 /* Transport methods defined outside transport.c */
156 int transport_helper_init(struct transport *transport, const char *name);
157 +int bidirectional_transfer_loop(int input, int output);
158
159 /* common methods used by transport.c and builtin-send-pack.c */
160 void transport_verify_remote_names(int nr_heads, const char **heads);