fhem.pl/01_FHEMWEB.pm/TcpServerUtils.pm: Nonblocking patches by geek (Forum #24799)

git-svn-id: https://svn.fhem.de/fhem/trunk@7212 2b470e98-0d58-463d-a4d8-8e2adae1ed80

fhem/FHEM/01_FHEMWEB.pm

@@ -7,6 +7,7 @@ use warnings;
 use TcpServerUtils;
 use HttpUtils;
 use Time::HiRes qw(gettimeofday);
+use Errno qw(:POSIX);
 
 #########################
 # Forward declaration
@@ -27,7 +28,6 @@ sub FW_makeEdit($$$);
 sub FW_makeImage(@);
 sub FW_makeTable($$$@);
 sub FW_makeTableFromArray($$@);
-sub FW_myPrint($$);
 sub FW_pF($@);
 sub FW_pH(@);
 sub FW_pHPlain(@);
@@ -175,7 +175,7 @@ FHEMWEB_Initialize($)
   # Initialize internal structures
   map { addToAttrList($_) } ( "webCmd", "icon", "devStateIcon",
                               "widgetOverride",  "sortby", "devStateStyle");
-  InternalTimer(time()+60, "FW_closeOldClients", 0, 0);
+  InternalTimer(time()+60, "FW_closeInactiveClients", 0, 0);
 
   $FW_dir      = "$attr{global}{modpath}/www";
   $FW_icondir  = "$FW_dir/images";
@@ -261,8 +261,10 @@ FW_Read($)
 
   if($hash->{SERVERSOCKET}) {   # Accept and create a child
     my $nhash = TcpServer_Accept($hash, "FHEMWEB");
+    return if(!$nhash);
     my $wt = AttrVal($name, "alarmTimeout", undef);
     $nhash->{ALARMTIMEOUT} = $wt if($wt);
+    $nhash->{CD}->blocking(0);
     return;
   }
 
@@ -292,14 +294,19 @@ FW_Read($)
   my $buf;
   my $ret = sysread($c, $buf, 1024);
 
-  if(!defined($ret) || $ret <= 0) {
+  if(!defined($ret) && $! == EWOULDBLOCK ){
+    $hash->{wantWrite} = 1
+      if(TcpServer_WantWrite($hash));
+    return;
+  } elsif(!$ret) { # 0==EOF, undef=error
     CommandDelete(undef, $name);
-    Log3 $FW_wname, 4, "Connection closed for $name";
+    Log3 $FW_wname, 4, "Connection closed for $name: ".
+                (defined($ret) ? 'EOF' : $!);
     return;
   }
 
   $hash->{BUF} .= $buf;
-  if($defs{$FW_wname}{SSL} && $c->can('pending')) {
+  if($hash->{SSL} && $c->can('pending')) {
     while($c->pending()) {
       sysread($c, $buf, 1024);
       $hash->{BUF} .= $buf;
@@ -351,20 +358,21 @@ FW_Read($)
       }
     }
     if($headerOptions[0]) {
-      print $c "HTTP/1.1 200 OK\r\n",
+      TcpServer_WriteBlocking($hash,
-             $FW_headercors,
+             "HTTP/1.1 200 OK\r\n".
-             "Content-Length: 0\r\n\r\n";
+             $FW_headercors.
+             "Content-Length: 0\r\n\r\n");
       delete $hash->{CONTENT_LENGTH};
       delete $hash->{BUF};
       return;
-      exit(1);
     };
     if(!$pwok) {
       my $msg = AttrVal($FW_wname, "basicAuthMsg", "Fhem: login required");
-      print $c "HTTP/1.1 401 Authorization Required\r\n",
+      TcpServer_WriteBlocking($hash,
-             "WWW-Authenticate: Basic realm=\"$msg\"\r\n",
+             "HTTP/1.1 401 Authorization Required\r\n".
-             $FW_headercors,
+             "WWW-Authenticate: Basic realm=\"$msg\"\r\n".
-             "Content-Length: 0\r\n\r\n";
+             $FW_headercors.
+             "Content-Length: 0\r\n\r\n");
       delete $hash->{CONTENT_LENGTH};
       delete $hash->{BUF};
       return;
@@ -383,22 +391,38 @@ FW_Read($)
   $arg = "" if(!defined($arg));
   Log3 $FW_wname, 4, "HTTP $name GET $arg";
   $FW_ME = "/" . AttrVal($FW_wname, "webname", "fhem");
-  my $pid;
   my $pf = AttrVal($FW_wname, "plotfork", undef);
   if($pf) {   # 0 disables
     # Process SVG rendering as a parallel process
     my $p = $data{FWEXT};
     if(grep { $p->{$_}{FORKABLE} && $arg =~ m+^$FW_ME$_+ } keys %{$p}) {
-      if($pid = fork) {
+      my $pid = fork();
+      if($pid) { # success, parent
 	use constant PRIO_PROCESS => 0;
 	setpriority(PRIO_PROCESS, $pid, getpriority(PRIO_PROCESS,$pid) + $pf);
+        # a) while child writes a new request might arrive if client uses
+        # pipelining or
+        # b) parent doesn't know about ssl-session changes due to child writing
+        # to socket
+        # -> have to close socket in parent... so that its only used in this
+        # child.
+	TcpServer_Disown( $hash );
+	delete($defs{$name});
 	return;
-      }
+
+      } elsif(defined($pid)){ # child
+	$hash->{isChild} = 1;
+
+      } # fork failed and continue in parent
     }
   }
 
   my $cacheable = FW_answerCall($arg);
-  return if($cacheable == -1); # Longpoll / inform request;
+  if($cacheable == -1){
+    # Longpoll / inform request;
+    exit if($hash->{isChild});
+    return;
+  }
 
   my $compressed = "";
   if(($FW_RETTYPE =~ m/text/i ||
@@ -413,14 +437,20 @@ FW_Read($)
   my $length = length($FW_RET);
   my $expires = ($cacheable?
                         ("Expires: ".localtime($now+900)." GMT\r\n") : "");
-  Log3 $FW_wname, 4, "$arg / RL:$length / $FW_RETTYPE / $compressed / $expires";
+  Log3 $FW_wname, 4,
-  $hash->{pid} = $pid if(defined($pid));
+        "$$:$name: $arg / RL:$length / $FW_RETTYPE / $compressed / $expires";
-  addToWritebuffer($hash, 
+  if( ! addToWritebuffer($hash,
            "HTTP/1.1 200 OK\r\n" .
            "Content-Length: $length\r\n" .
            $expires . $compressed . $FW_headercors .
            "Content-Type: $FW_RETTYPE\r\n\r\n" .
-           $FW_RET, "FW_closeConn");
+           $FW_RET, "FW_closeConn") ){
+    Log3 $name, 4, "Closing connection $name due to full buffer in FW_Read";
+    TcpServer_Close( $hash );
+    delete($defs{$name});
+  }
+
+  exit if($hash->{isChild});
 }
 
 sub
@@ -432,7 +462,7 @@ FW_closeConn($)
     TcpServer_Close($hash);
     delete($defs{$hash->{NAME}});
   }
-  exit if(defined($hash->{pid}));
+  exit if($hash->{isChild});
 }
 
 ###########################
@@ -509,11 +539,12 @@ FW_answerCall($)
     $arg = $1; # The stuff behind FW_ME, continue to check for commands/FWEXT
 
   } else {
-    my $c = $me->{CD};
     Log3 $FW_wname, 4, "$FW_wname: redirecting $arg to $FW_ME";
-    print $c "HTTP/1.1 302 Found\r\n",
+    TcpServer_WriteBlocking($me,
-             "Content-Length: 0\r\n", $FW_headercors,
+             "HTTP/1.1 302 Found\r\n".
-             "Location: $FW_ME\r\n\r\n";
+             "Content-Length: 0\r\n".
+             $FW_headercors.
+             "Location: $FW_ME\r\n\r\n");
     FW_closeConn($FW_chash);
     return -1;
   }
@@ -555,11 +586,11 @@ FW_answerCall($)
     $me->{NTFY_ORDER} = $FW_cname;   # else notifyfn won't be called
     %ntfyHash = ();
 
-    my $c = $me->{CD};
+    TcpServer_WriteBlocking($me,
-    print $c "HTTP/1.1 200 OK\r\n",
+       "HTTP/1.1 200 OK\r\n".
-       $FW_headercors,
+       $FW_headercors.
-       "Content-Type: application/octet-stream; charset=$FW_encoding\r\n\r\n",
+       "Content-Type: application/octet-stream; charset=$FW_encoding\r\n\r\n".
-       FW_roomStatesForInform($me);
+       FW_roomStatesForInform($me));
     return -1;
   }
 
@@ -619,11 +650,11 @@ FW_answerCall($)
     my $tgt = $FW_ME;
        if($FW_detail) { $tgt .= "?detail=$FW_detail" }
     elsif($FW_room)   { $tgt .= "?room=$FW_room" }
-    my $c = $me->{CD};
+    TcpServer_WriteBlocking($me,
-    print $c "HTTP/1.1 302 Found\r\n",
+             "HTTP/1.1 302 Found\r\n".
-             "Content-Length: 0\r\n", $FW_headercors,
+             "Content-Length: 0\r\n". $FW_headercors.
-             "Location: $tgt\r\n",
+             "Location: $tgt\r\n".
-             "\r\n";
+             "\r\n");
     return -1;
   }
 
@@ -1438,20 +1469,10 @@ FW_fileList($)
 sub
 FW_outputChunk($$$)
 {
-  my ($c, $buf, $d) = @_;
+  my ($hash, $buf, $d) = @_;
   $buf = $d->deflate($buf) if($d);
-  FW_myPrint($c, sprintf("%x\r\n", length($buf)).$buf."\r\n") if(length($buf));
+  if( length($buf) ){
-}
+    TcpServer_WriteBlocking($hash, sprintf("%x\r\n",length($buf)) .$buf."\r\n");
-
-sub
-FW_myPrint($$)
-{
-  my ($c, $buf) = @_;
-  my ($off, $len) = (0, length($buf));
-  while($off < $len) {
-    my $ret = syswrite($c, $buf, $len-$off, $off);
-    last if(!$ret || $ret < 0);
-    $off += $ret;
   }
 }
 
@@ -1462,7 +1483,6 @@ FW_returnFileAsStream($$$$$)
 
   my $etag;
 
-  my $c = $FW_chash->{CD};
   if($cacheable) {
     #Check for If-None-Match header (ETag)
     my @if_none_match_lines = grep /If-None-Match/, @FW_httpheader;
@@ -1474,7 +1494,7 @@ FW_returnFileAsStream($$$$$)
 
     $etag = (stat($path))[9]; #mtime
     if(defined($etag) && defined($if_none_match) && $etag eq $if_none_match) {
-      FW_myPrint($c,"HTTP/1.1 304 Not Modified\r\n".
+      TcpServer_WriteBlocking($FW_chash,"HTTP/1.1 304 Not Modified\r\n".
                     $FW_headercors . "\r\n");
       FW_closeConn($FW_chash);
       return -1;
@@ -1493,29 +1513,32 @@ FW_returnFileAsStream($$$$$)
   my $expires = $cacheable ? ("Expires: ".gmtime(time()+900)." GMT\r\n"): "";
   my $compr = ((int(@FW_enc) == 1 && $FW_enc[0] =~ m/gzip/) && $FW_use_zlib) ?
                 "Content-Encoding: gzip\r\n" : "";
-  FW_myPrint($c, "HTTP/1.1 200 OK\r\n".
+  TcpServer_WriteBlocking($FW_chash, "HTTP/1.1 200 OK\r\n".
                   $compr . $expires . $FW_headercors . $etag .
                   "Transfer-Encoding: chunked\r\n" .
                   "Content-Type: $type; charset=$FW_encoding\r\n\r\n");
 
   my $d = Compress::Zlib::deflateInit(-WindowBits=>31) if($compr);
-  FW_outputChunk($c, $FW_RET, $d);
+  FW_outputChunk($FW_chash, $FW_RET, $d);
   my $buf;
   while(sysread(FH, $buf, 2048)) {
     if($doEsc) { # FileLog special
       $buf =~ s/</&lt;/g;
       $buf =~ s/>/&gt;/g;
     }
-    FW_outputChunk($c, $buf, $d);
+    FW_outputChunk($FW_chash, $buf, $d);
   }
   close(FH);
-  FW_outputChunk($c, $suffix, $d);
+  FW_outputChunk($FW_chash, $suffix, $d);
 
   if($compr) {
     $buf = $d->flush();
-    FW_myPrint($c,sprintf("%x\r\n",length($buf)).$buf."\r\n") if($buf);
+    if($buf){
+      TcpServer_WriteBlocking($FW_chash,
+                sprintf("%x\r\n",length($buf)) .$buf."\r\n");
+    }
   }
-  FW_myPrint($c, "0\r\n\r\n");
+  TcpServer_WriteBlocking($FW_chash, "0\r\n\r\n");
   FW_closeConn($FW_chash);
   return -1;
 }
@@ -2267,8 +2290,15 @@ FW_Notify($$)
     }
   }
 
-  addToWritebuffer($ntfy, join("\n", map { s/\n/ /gm; $_ } @data)."\n")
+  if(@data){
-    if(@data);
+    if(!addToWritebuffer($ntfy, join("\n", map { s/\n/ /gm; $_ } @data)."\n") ){
+      my $name = $ntfy->{NAME};
+      Log3 $name, 4, "Closing connection $name due to full buffer in FW_Notify";
+      TcpServer_Close($ntfy);
+      delete($defs{$name});
+    }
+  }
+
   return undef;
 }
 
@@ -2423,18 +2453,18 @@ FW_Set($@)
 
 #####################################
 sub
-FW_closeOldClients()
+FW_closeInactiveClients()
 {
   my $now = time();
   foreach my $dev (keys %defs) {
     next if(!$defs{$dev}{TYPE} || $defs{$dev}{TYPE} ne "FHEMWEB" ||
             !$defs{$dev}{LASTACCESS} || $defs{$dev}{inform} ||
             ($now - $defs{$dev}{LASTACCESS}) < 60);
-    Log3 $FW_wname, 4, "Closing connection $dev";
+    Log3 $FW_wname, 4, "Closing inactive connection $dev";
     FW_Undef($defs{$dev}, "");
     delete $defs{$dev};
   }
-  InternalTimer($now+60, "FW_closeOldClients", 0, 0);
+  InternalTimer($now+60, "FW_closeInactiveClients", 0, 0);
 }
 
 sub

fhem/FHEM/TcpServerUtils.pm

@@ -5,6 +5,7 @@ package main;
 use strict;
 use warnings;
 use IO::Socket;
+use Errno qw(:POSIX);
 
 sub
 TcpServer_Open($$$)
@@ -93,8 +94,12 @@ TcpServer_Accept($$)
       SSL_version   => 'SSLv23:!SSLv3:!SSLv2', #Forum #27565
       Timeout       => 4,
       });
-    if(!$ret && $! ne "Socket is not connected") {
+    my $err = $!;
-      Log3 $name, 1, "$type SSL/HTTPS error: $!";
+    if( !$ret
+      && $err != EWOULDBLOCK
+      && $err ne "Socket is not connected") {
+
+      Log3 $name, 1, "$type SSL/HTTPS error: $err";
       close($clientinfo[0]);
       return undef;
     }
@@ -107,6 +112,7 @@ TcpServer_Accept($$)
   $nhash{FD}    = $clientinfo[0]->fileno();
   $nhash{CD}    = $clientinfo[0];     # sysread / close won't work on fileno
   $nhash{TYPE}  = $type;
+  $nhash{SSL}   = $hash->{SSL};
   $nhash{STATE} = "Connected";
   $nhash{SNAME} = $name;
   $nhash{TEMPORARY} = 1;              # Don't want to save it
@@ -155,4 +161,118 @@ TcpServer_Close($)
   }
   return undef;
 }
+
+# close a (SSL-)Socket in local process
+# avoids interfering with other processes using it
+# this is critical for SSL and helps with other issues, too 
+sub
+TcpServer_Disown($)
+{
+  my ($hash) = @_;
+  my $name = $hash->{NAME};
+
+  if( defined($hash->{CD}) ){
+    if( $hash->{SSL} ){
+      $hash->{CD}->close( SSL_no_shutdown => 1);
+    } else {
+      close( $hash->{CD} );
+    }
+    delete($hash->{CD});
+    delete($selectlist{$name});
+    delete($hash->{FD});  # Avoid Read->Close->Write
+  }
+
+  return;
+}
+
+# wait for a socket to become ready
+# takes IO::Socket::SSL + non-blocking into account
+sub
+TcpServer_Wait($$)
+{
+  my( $hash, $direction ) = @_;
+
+  my $read = '';
+  my $write ='';
+
+  if( $direction eq 'read' || $hash->{wantRead} ){
+    vec( $read, $hash->{FD}, 1) = 1;
+  } elsif( $direction eq 'write' || $hash->{wantWrite} ){
+    vec( $write, $hash->{FD}, 1) = 1;
+  } else {
+    return undef;
+  }
+
+  my $ret = select( $read, $write, undef, undef );
+  return if $ret == -1;
+
+  if( vec( $read, $hash->{FD}, 1) ){
+    delete $hash->{wantRead};
+  }
+  if( vec( $write, $hash->{FD}, 1) ){
+    delete $hash->{wantWrite};
+  }
+
+  # return true on success
+  return 1;
+}
+
+# WantRead/Write: keep ssl constants local
+sub
+TcpServer_WantRead($)
+{
+  my( $hash ) = @_;
+
+  return $hash->{SSL}
+	&& $hash->{CD}
+	&& $hash->{CD}->errstr == &IO::Socket::SSL::SSL_WANT_READ;
+}
+
+sub
+TcpServer_WantWrite($)
+{
+  my( $hash ) = @_;
+
+  return $hash->{SSL}
+	&& $hash->{CD}
+	&& $hash->{CD}->errstr == &IO::Socket::SSL::SSL_WANT_WRITE;
+}
+
+# write until all data is done.
+# hanldes both, blocking and non-blocking sockets
+# ... with or without SSL
+sub
+TcpServer_WriteBlocking($$)
+{
+  my( $hash, $txt ) = @_;
+
+  my $sock = $hash->{CD};
+  return undef if(!$sock);
+  my $off = 0;
+  my $len = length($txt);
+
+  while($off < $len) {
+    if(!TcpServer_Wait($hash, 'write')) {
+      TcpServer_Close($hash);
+      return undef;
+    }
+
+    my $ret = syswrite($sock, $txt, $len-$off, $off);
+
+    if( defined $ret ){
+      $off += $ret;
+
+    } elsif( $! == EWOULDBLOCK ){
+      $hash->{wantRead} = 1
+        if TcpServer_WantRead($hash);
+
+    } else {
+      TcpServer_Close($hash);
+      return undef; # error
+    }
+  }
+
+  return 1; # success
+}
+
 1;

fhem/fhem.pl

@@ -34,6 +34,7 @@ use strict;
 use warnings;
 use IO::Socket;
 use Time::HiRes qw(gettimeofday);
+use Errno qw(:POSIX);
 
 
 ##################################################
@@ -541,9 +542,11 @@ while (1) {
     my $hash = $selectlist{$p};
     if(defined($hash->{FD})) {
       vec($rin, $hash->{FD}, 1) = 1
-        if(!defined($hash->{directWriteFn}));
+        if(!defined($hash->{directWriteFn}) && !$hash->{wantWrite} );
       vec($win, $hash->{FD}, 1) = 1
-        if(defined($hash->{directWriteFn}) || defined($hash->{$wbName}));
+        if( (defined($hash->{directWriteFn}) ||
+             defined($hash->{$wbName}) || 
+             $hash->{wantWrite} ) && !$hash->{wantRead} );
     }
     vec($ein, $hash->{EXCEPT_FD}, 1) = 1
         if(defined($hash->{"EXCEPT_FD"}));
@@ -595,6 +598,8 @@ while (1) {
     next if(!$isDev && !$isDirect);
 
     if(defined($hash->{FD}) && vec($rout, $hash->{FD}, 1)) {
+      delete $hash->{wantRead};
+
       if($hash->{directReadFn}) {
         $hash->{directReadFn}($hash);
       } else {
@@ -602,21 +607,28 @@ while (1) {
       }
     }
 
-    if((defined($hash->{$wbName}) || defined($hash->{directWriteFn})) &&
+    if( defined($hash->{FD}) && vec($wout, $hash->{FD}, 1)) {
-        defined($hash->{FD}) && vec($wout, $hash->{FD}, 1)) {
+      delete $hash->{wantWrite};
 
       if($hash->{directWriteFn}) {
         $hash->{directWriteFn}($hash);
 
-      } else {
+      } elsif(defined($hash->{$wbName})) {
         my $wb = $hash->{$wbName};
         alarm($hash->{ALARMTIMEOUT}) if($hash->{ALARMTIMEOUT});
         my $ret = syswrite($hash->{CD}, $wb);
+        my $werr = int($!);
         alarm(0) if($hash->{ALARMTIMEOUT});
-        if(!$ret || $ret < 0) {
+
+        if(!defined($ret) && $werr == EWOULDBLOCK ) {
+	  $hash->{wantRead} = 1
+	    if(TcpServer_WantRead($hash));
+
+        } elsif(!$ret) { # zero=EOF, undef=error
           Log 4, "Write error to $p, deleting $hash->{NAME}";
           TcpServer_Close($hash);
           CommandDelete(undef, $hash->{NAME});
+
         } else {
           if($ret == length($wb)) {
             delete($hash->{$wbName});
@@ -3855,13 +3867,8 @@ addToWritebuffer($$@)
 {
   my ($hash, $txt, $callback, $nolimit) = @_;
 
-  if(defined($hash->{pid})) {  # Wont go to the main select in a forked process
+  if($hash->{isChild}) {  # Wont go to the main select in a forked process
-    my ($off, $len) = (0, length($txt));
+    TcpServer_WriteBlocking( $hash, $txt );
-    while($off < $len) {
-      my $ret = syswrite($hash->{CD}, $txt, $len-$off, $off);
-      last if(!$ret || $ret <= 0);
-      $off += $ret;
-    }
     if($callback) {
       no strict "refs";
       my $ret = &{$callback}($hash);
@@ -3875,7 +3882,11 @@ addToWritebuffer($$@)
     $hash->{$wbName} = $txt;
   } elsif($nolimit || length($hash->{$wbName}) < 102400) {
     $hash->{$wbName} .= $txt;
+  } else {
+    return 0;
   }
+
+  return 1; # success
 }
 
 sub