Don't flush pinned pages in checkpoint (fix for my last push)

storage/maria/unittest/ma_pagecache_single.c:
  Check if we can flush write locked page for normal flush
  Check that we can't flush read locked page with FLUSH_KEEP_LAZY (used by checkpoint)

storage/maria/ma_pagecache.c

@@ -4188,12 +4188,15 @@ static int flush_cached_blocks(PAGECACHE *pagecache,
     PAGECACHE_BLOCK_LINK *block= *cache;
 
     /*
-      This code is only run for non transactional tables.
+      In the case of non_transactional tables we want to flush also
-      We may have other threads reading the block during flush,
+      block pinned with reads. This is becasue we may have other
-      as non transactional tables can have many readers while the
+      threads reading the block during flush, as non transactional
-      one writer is doing the flush.
+      tables can have many readers while the one writer is doing the
+      flush.
+      We don't want to do flush pinned blocks during checkpoint.
+      We detect the checkpoint case by checking if type is LAZY.
     */
-    if (block->wlocks)
+    if ((type == FLUSH_KEEP_LAZY && block->pins) || block->wlocks)
     {
       KEYCACHE_DBUG_PRINT("flush_cached_blocks",
                           ("block: %u (0x%lx)  pinned",

storage/maria/unittest/ma_pagecache_single.c

@@ -246,7 +246,7 @@ int simple_read_change_write_read_test()
 
 /*
   Prepare page, read page 0 (and pin) then write page 1 and page 0.
-  Flush the file (shold flush only page 1 and return 1 (page 0 is
+  Flush the file (should flush only page 1 and return 1 (page 0 is
   still pinned).
   Check file on the disk.
   Unpin and flush.
@@ -284,7 +284,7 @@ int simple_pin_test()
   bfill(buffw + TEST_PAGE_SIZE/2, TEST_PAGE_SIZE/2, ((unsigned char) 129));
   pagecache_write(&pagecache, &file1, 0, 3, buffw,
                   PAGECACHE_PLAIN_PAGE,
-                  PAGECACHE_LOCK_WRITE_TO_READ,
+                  PAGECACHE_LOCK_LEFT_WRITELOCKED,
                   PAGECACHE_PIN_LEFT_PINNED,
                   PAGECACHE_WRITE_DELAY,
                   0, LSN_IMPOSSIBLE);
@@ -304,7 +304,7 @@ int simple_pin_test()
   pagecache_unlock(&pagecache,
                    &file1,
                    0,
-                   PAGECACHE_LOCK_READ_UNLOCK,
+                   PAGECACHE_LOCK_WRITE_UNLOCK,
                    PAGECACHE_UNPIN,
                    0, 0, 0);
   if (flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE))
@@ -323,6 +323,93 @@ err:
   DBUG_RETURN(res);
 }
 
+/*
+  Prepare page, read page 0 (and pin) then write page 1 and page 0.
+  Flush the file (should flush only page 1 and return 1 (page 0 is
+  still pinned).
+  Check file on the disk.
+  Unpin and flush.
+  Check file on the disk.
+*/
+int simple_pin_test2()
+{
+  unsigned char *buffw= malloc(TEST_PAGE_SIZE);
+  int res;
+  DBUG_ENTER("simple_pin_test2");
+  /* prepare the file */
+  bfill(buffw, TEST_PAGE_SIZE, '\1');
+  pagecache_write(&pagecache, &file1, 0, 3, buffw,
+                  PAGECACHE_PLAIN_PAGE,
+                  PAGECACHE_LOCK_LEFT_UNLOCKED,
+                  PAGECACHE_PIN_LEFT_UNPINNED,
+                  PAGECACHE_WRITE_DELAY,
+                  0, LSN_IMPOSSIBLE);
+  /* test */
+  if (flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE))
+  {
+    diag("Got error during flushing pagecache\n");
+    exit(1);
+  }
+  pagecache_read(&pagecache, &file1, 0, 3, buffw,
+                 PAGECACHE_PLAIN_PAGE,
+                 PAGECACHE_LOCK_WRITE,
+                 0);
+  pagecache_write(&pagecache, &file1, 1, 3, buffw,
+                  PAGECACHE_PLAIN_PAGE,
+                  PAGECACHE_LOCK_LEFT_UNLOCKED,
+                  PAGECACHE_PIN_LEFT_UNPINNED,
+                  PAGECACHE_WRITE_DELAY,
+                  0, LSN_IMPOSSIBLE);
+  bfill(buffw + TEST_PAGE_SIZE/2, TEST_PAGE_SIZE/2, ((unsigned char) 129));
+  pagecache_write(&pagecache, &file1, 0, 3, buffw,
+                  PAGECACHE_PLAIN_PAGE,
+                  PAGECACHE_LOCK_WRITE_TO_READ,
+                  PAGECACHE_PIN_LEFT_PINNED,
+                  PAGECACHE_WRITE_DELAY,
+                  0, LSN_IMPOSSIBLE);
+  /*
+    We have to get error because one page of the file is pinned,
+    other page should be flushed
+  */
+  if (!flush_pagecache_blocks(&pagecache, &file1, FLUSH_KEEP_LAZY))
+  {
+    diag("Did not get error in flush_pagecache_blocks 2\n");
+    res= 0;
+    goto err;
+  }
+  ok((res= test(test_file(file1, file1_name, TEST_PAGE_SIZE*2, TEST_PAGE_SIZE*2,
+                           simple_pin_test_file1))),
+     "Simple pin page file with pin 2");
+
+  /* Test that a normal flush goes through */
+  if (flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE))
+  {
+    diag("Got error in flush_pagecache_blocks 3\n");
+    res= 0;
+    goto err;
+  }
+  pagecache_unlock(&pagecache,
+                   &file1,
+                   0,
+                   PAGECACHE_LOCK_READ_UNLOCK,
+                   PAGECACHE_UNPIN,
+                   0, 0, 0);
+  if (flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE))
+  {
+    diag("Got error in flush_pagecache_blocks 4\n");
+    res= 0;
+    goto err;
+  }
+  ok((res&= test(test_file(file1, file1_name, TEST_PAGE_SIZE*2, TEST_PAGE_SIZE,
+                           simple_pin_test_file2))),
+     "Simple pin page result file 2");
+  if (res)
+    reset_file(&file1, file1_name);
+err:
+  free(buffw);
+  DBUG_RETURN(res);
+}
+
 /*
   Checks pins without lock.
 */
@@ -357,7 +444,7 @@ int simple_pin_no_lock_test()
     We have to get error because one page of the file is pinned,
     other page should be flushed
   */
-  if (!flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE))
+  if (!flush_pagecache_blocks(&pagecache, &file1, FLUSH_KEEP_LAZY))
   {
     diag("Did not get error in flush_pagecache_blocks 2\n");
     res= 0;
@@ -392,7 +479,7 @@ int simple_pin_no_lock_test()
   pagecache_unlock_by_link(&pagecache, link,
                            PAGECACHE_LOCK_WRITE_UNLOCK,
                            PAGECACHE_PIN_LEFT_PINNED, 0, 0, 1, FALSE);
-  if (!flush_pagecache_blocks(&pagecache, &file1, FLUSH_FORCE_WRITE))
+  if (!flush_pagecache_blocks(&pagecache, &file1, FLUSH_KEEP_LAZY))
   {
     diag("Did not get error in flush_pagecache_blocks 3\n");
     res= 0;
@@ -609,6 +696,7 @@ static void *test_thread(void *arg)
   if (!simple_read_write_test() ||
       !simple_read_change_write_read_test() ||
       !simple_pin_test() ||
+      !simple_pin_test2() ||
       !simple_pin_no_lock_test() ||
       !simple_delete_forget_test() ||
       !simple_delete_flush_test())
@@ -657,8 +745,8 @@ int main(int argc __attribute__((unused)),
   DBUG_ENTER("main");
   DBUG_PRINT("info", ("Main thread: %s\n", my_thread_name()));
 
-  plan(16);
+  plan(18);
-  SKIP_BIG_TESTS(16)
+  SKIP_BIG_TESTS(18)
   {
 
   if ((tmp_file= my_open(file2_name, O_CREAT | O_TRUNC | O_RDWR,