WvStreams
wvunixdgsocket.cc
00001 #include "wvunixdgsocket.h"
00002 
00003 WvUnixDGSocket::WvUnixDGSocket(WvStringParm filename, bool _server, int perms)
00004     : socketfile(filename)
00005 {
00006 //    log(WvLog::Debug2, "Starting up %s!\n", filename);
00007     server = _server;
00008     backoff = 10;
00009 
00010     bufsize = 0;
00011 
00012     // open a datagram unix domain socket
00013     setfd(socket(PF_UNIX, SOCK_DGRAM, 0));
00014 
00015     // if we don't have a file desciptor, something is wrong.
00016     if (getfd() < 0)
00017     {
00018         seterr("No Socket available.");
00019         return;
00020     }
00021 
00022     // set non-blocking mode
00023     fcntl(getfd(), F_SETFL, O_RDWR|O_NONBLOCK);
00024 
00025     WvUnixAddr uaddr(socketfile);
00026 
00027     // Let this file be reusable, since we're going to own this anyway
00028     // The business with the int x is just Unix stupidities.. *sigh*
00029     int x = 1;
00030     setsockopt(getfd(), SOL_SOCKET, SO_REUSEADDR, &x, sizeof(x));
00031 
00032     if (server)
00033     {
00034         // Fix it so that there can't be another process on this file
00035         unlink(socketfile);
00036 
00037         // Actually bind to the address we set up above.
00038         sockaddr *addr = uaddr.sockaddr();
00039         if (bind(getfd(), (sockaddr *)addr, uaddr.sockaddr_len()))
00040         {
00041             seterr("Bind to %s failed: %s", socketfile, strerror(errno));
00042             close();
00043         }
00044         delete addr;
00045 
00046         chmod(socketfile, perms);
00047     }
00048     else
00049     {
00050         // we're the client, so we connect to someone else's socket
00051         sockaddr *addr = uaddr.sockaddr();
00052         if (connect(getfd(), (sockaddr *)addr, uaddr.sockaddr_len()))
00053         {
00054             seterr("Connect to %s failed: %s",
00055                    socketfile, strerror(errno));
00056             close();
00057         }
00058         delete addr;
00059     }
00060 
00061     drain();
00062 }
00063 
00064 WvUnixDGSocket::~WvUnixDGSocket()
00065 {
00066 //    log(WvLog::Debug2, "Destroying: %s\n", socketfile);
00067     close();
00068     if (server)
00069         unlink(socketfile);
00070 }
00071 
00072 size_t WvUnixDGSocket::uwrite(const void *buf, size_t count)
00073 {
00074     size_t ret = bufs.isempty() ? WvFDStream::uwrite(buf, count) : 0;
00075 
00076     if (ret < count)
00077     {
00078         WvDynBuf *b = new WvDynBuf;
00079         b->put(buf, count);
00080         bufs.append(b, true);
00081         bufsize += count;
00082     }
00083 
00084     return count;
00085 }
00086 
00087 void WvUnixDGSocket::pre_select(SelectInfo &si)
00088 {
00089     SelectRequest oldwant = si.wants;
00090     if (!bufs.isempty())
00091     {
00092         // stupid unix domain sockets seem to return true when selecting
00093         // for write EVEN IF write() RETURNS -EAGAIN!  Just shoot me.
00094         // 
00095         // To deal with this, we set an alarm() in post_select() if we
00096         // couldn't write everything we wanted.  While the alarm is set,
00097         // we don't try to flush our output buffer.
00098         if (alarm_remaining() <= 0)
00099             si.wants.writable = true;
00100         else if (si.msec_timeout < 0
00101                  || si.msec_timeout > alarm_remaining())
00102             si.msec_timeout = alarm_remaining();
00103     }
00104 
00105     WvFDStream::pre_select(si);
00106 
00107     si.wants = oldwant;
00108 }
00109 
00110 bool WvUnixDGSocket::post_select(SelectInfo &si)
00111 {
00112     SelectRequest oldwant = si.wants;
00113     if (!bufs.isempty())
00114         si.wants.writable = true;
00115 
00116     bool sure = WvFDStream::post_select(si);
00117 
00118     si.wants = oldwant;
00119 
00120     if (sure)
00121     {
00122         // try flushing previous bufs
00123         WvBufList::Iter i(bufs);
00124         for (i.rewind(); i.next(); )
00125         {
00126             int used = i->used();
00127             int retval = WvFDStream::uwrite(i->get(used), used);
00128             if (retval < used)
00129             {
00130                 i->unget(used);
00131                 alarm(backoff *= 2);
00132                 if (backoff > 1000)
00133                     backoff = 1000;
00134                 break; // can't continue
00135             }
00136             else
00137             {
00138                 bufsize -= used;
00139                 i.xunlink(); // done with that one
00140                 backoff = 10;
00141             }
00142         }
00143     }
00144 
00145     return sure;
00146 }
00147 
00148