1 #ifndef DUMP_CONNECTION_PIPE_H
  2 #define DUMP_CONNECTION_PIPE_H
  3 
  4 #include <type_traits>
  5 #include <unistd.h>
  6 #include <stdint.h>
  7 #include <string.h>
  8 #include <stdio.h>
  9 #include <stdlib.h>
 10 
 11 
 12 enum ConnPermission {
 13     CONN_READABLE = 1,
 14     CONN_WRITABLE = 2
 15 };
 16 
 17 template<ConnPermission permission>
 18 class ConnectionPipe
 19 {
 20 private:
 21     //use for multi process atomic write one pipe
 22     #pragma pack(1)
 23     struct AtomicPipeData {
 24         static const size_t max_cap = 256;
 25         char data[max_cap];
 26         uint16_t size:14;
 27         uint16_t status:2; //1 is sending, 0 is end
 28         //uint64_t id; use for package join to msg
 29     };
 30     #pragma pack()
 31 public:
 32     explicit ConnectionPipe(int fd) : fd_(fd){}
 33     ConnectionPipe(const ConnectionPipe&) = delete;
 34     ConnectionPipe& operator=(const ConnectionPipe&) = delete;
 35     ~ConnectionPipe() = default;
 36 
 37     void Close(){
 38         close(fd_);
 39     }
 40 
 41     int FileNo() const {
 42         return fd_;
 43     }
 44 
 45     //bool Poll();
 46 
 47     //int Recv();
 48     char* RecvBytes(size_t* msg_len);
 49 
 50     //int RecvBytesInto();
 51     //bool Send();
 52     bool SendBytes(const char* msg, size_t msg_len);
 53 private:
 54     int fd_;
 55 };
 56 
 57 template<>
 58 char* ConnectionPipe<CONN_READABLE>::RecvBytes(size_t* msg_len) {
 59     if (msg_len == nullptr) {
 60         return nullptr;
 61     }
 62 
 63     int nreaded = 0;
 64     AtomicPipeData buf = {0};
 65     size_t msg_cap = AtomicPipeData::max_cap * 16; //4096 one page
 66     char * msg = new char[msg_cap];
 67     size_t msg_offset = 0;
 68     auto resize_mgs = [&msg, &msg_cap, &msg_offset](){
 69         size_t recommended_cap = msg_cap * 2;
 70         char* new_msg = new char[recommended_cap];
 71         memset(new_msg, 0, recommended_cap);
 72         memcpy(new_msg, msg, msg_offset);
 73         delete[] msg;
 74         msg = new_msg;
 75         msg_cap = recommended_cap;
 76     };
 77 
 78     while (1) {
 79         int nread = read(fd_, (char*)&buf + nreaded, sizeof(buf) - nreaded);
 80         {
 81             if (nread == -1) {
 82                 perror("read");
 83                 exit(EXIT_FAILURE);
 84             }
 85             //child close write pipe
 86             if (nread == 0) break;
 87         }
 88 
 89         nreaded += nread;
 90         if (nreaded == sizeof(buf)) {
 91             if ((msg_offset + buf.size) > msg_cap) {
 92                 resize_mgs();
 93             }
 94 
 95             memcpy(msg + msg_offset, buf.data, buf.size);
 96             msg_offset += buf.size;
 97 
 98             //use for ending package buf of one message
 99             if (buf.status == 0) {
100                 break;
101             }
102 
103             //use for loop next step read
104             nreaded = 0;
105             memset(&buf, 0, sizeof(buf));
106         }
107     };
108 
109     *msg_len = msg_offset;
110     return msg;
111 }
112 
113 template<>
114 char* ConnectionPipe<CONN_WRITABLE>::RecvBytes(size_t* msg_len) {
115     return nullptr;
116 }
117 
118 template<>
119 bool ConnectionPipe<CONN_WRITABLE>::SendBytes(const char* msg, size_t msg_len) {
120     if (msg == nullptr || msg_len == 0) {
121         return false;
122     }
123 
124     AtomicPipeData package;
125     auto fill_package = [&package](const char * data, uint16_t len, uint16_t status) {
126         memset(&package, 0, sizeof(package));
127         memcpy(package.data, data, len);
128         package.size = len;
129         package.status = status;
130     };
131 
132     const size_t packages_size = (msg_len + AtomicPipeData::max_cap - 1)/AtomicPipeData::max_cap;
133     for (size_t i = 0; i < packages_size; ++i) {
134         size_t offset = i * AtomicPipeData::max_cap;
135         if (i != packages_size - 1) {
136             fill_package(msg+offset, AtomicPipeData::max_cap, 1);
137         }
138         else {
139             fill_package(msg+offset, msg_len - offset, 0);
140         }
141 
142         if (write(fd_, &package, sizeof(package)) != sizeof(package)) {
143             /* Nothing to do on error, this will be detected by the other side. */
144         }
145     }
146 
147     return true ;
148 }
149 
150 template<>
151 bool ConnectionPipe<CONN_READABLE>::SendBytes(const char* msg, size_t msg_len) {
152     return false;
153 }
154 
155 static bool NewConnPair(ConnectionPipe<CONN_WRITABLE>** pp_sender, ConnectionPipe<CONN_READABLE>** pp_receiver) {
156     int pipedes[2];
157     if (pipe(pipedes) == -1) {
158         close(pipedes[0]);
159         close(pipedes[1]);
160         return false;
161     }
162 
163     *pp_receiver = new ConnectionPipe<CONN_READABLE>(pipedes[0]);
164     *pp_sender   = new ConnectionPipe<CONN_WRITABLE>(pipedes[1]);
165 
166 
167     printf("pipe buffer size = %ld\n", fpathconf(pipedes[0], _PC_PIPE_BUF));
168     return true;
169 }
170 
171 #endif
172 
173 
174 int main()
175 {
176 
177    ConnectionPipe<CONN_WRITABLE>* p_sender   = nullptr;
178    ConnectionPipe<CONN_READABLE>* p_receiver = nullptr;
179    if (!NewConnPair(&p_sender, &p_receiver)) {
180         exit(EXIT_FAILURE);
181    }
182 
183    std::shared_ptr<ConnectionPipe<CONN_WRITABLE>> sender_ptr(p_sender);
184    std::shared_ptr<ConnectionPipe<CONN_READABLE>> receiver_ptr(p_receiver);
185 
186    pid_t regionid_pid = fork();
187    if (regionid_pid == -1) {
188        perror("fork");
189        exit(EXIT_FAILURE);
190    }
191    else if (regionid_pid == 0) {
192        //sub process
193        receiver_ptr->Close();
194        sender_ptr->SendBytes("bufff", 5);
195        sender_ptr->Close();
196        _exit(EXIT_SUCCESS);
197    }
198    else {
199        //parent mian process
200        sender_ptr->Close();
201        size_t msg_len = 0;
202        char* msg = receiver_ptr->RecvBytes(&msg_len);
203        receiver_ptr->Close();
204        waitpid(regionid_pid, NULL, 0);
205    }
206     
207     return 0
208 }
209